shorten default timeout for aggregate; allow configuration of timeout for aggregates

docs/api.md

@@ -195,10 +195,13 @@ Following is the supported API format for specifying metrics aggregations:
 
 <pre>
  aggregates:
-         name: description of aggregation result
-         groupByKeys: list of fields on which to aggregate
-         operationType: sum, min, max, count, avg or raw_values
-         operationKey: internal field on which to perform the operation
+         defaultExpiryTime: default time duration of data aggregation to perform rules (default: 2 minutes)
+         rules: list of aggregation rules, each includes:
+                 name: description of aggregation result
+                 groupByKeys: list of fields on which to aggregate
+                 operationType: sum, min, max, count, avg or raw_values
+                 operationKey: internal field on which to perform the operation
+                 expiryTime: time interval over which to perform the operation
 </pre>
 ## Connection tracking API
 Following is the supported API format for specifying connection tracking:

network_definitions/bandwidth_per_network_service.yaml

@@ -17,13 +17,15 @@ transform:
       type: add_service
       parameters: proto
 extract:
+  type: aggregates
   aggregates:
-    - name: bandwidth_network_service
-      groupByKeys:
-        - service
-        - _RecordType
-      operationType: sum
-      operationKey: bytes
+    rules:
+      - name: bandwidth_network_service
+        groupByKeys:
+          - service
+          - _RecordType
+        operationType: sum
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/bandwidth_per_src_dest_subnet.yaml

@@ -21,14 +21,16 @@ transform:
       type: add_subnet
       parameters: /24
 extract:
+  type: aggregates
   aggregates:
-    - name: bandwidth_source_destination_subnet
-      groupByKeys:
-        - dstSubnet24
-        - srcSubnet24
-        - _RecordType
-      operationType: sum
-      operationKey: bytes
+    rules:
+      - name: bandwidth_source_destination_subnet
+        groupByKeys:
+          - dstSubnet24
+          - srcSubnet24
+          - _RecordType
+        operationType: sum
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/bandwidth_per_src_subnet.yaml

@@ -17,13 +17,15 @@ transform:
       type: add_subnet
       parameters: /16
 extract:
+  type: aggregates
   aggregates:
-    - name: bandwidth_source_subnet
-      groupByKeys:
-        - srcSubnet
-        - _RecordType
-      operationType: sum
-      operationKey: bytes
+    rules:
+      - name: bandwidth_source_subnet
+        groupByKeys:
+          - srcSubnet
+          - _RecordType
+        operationType: sum
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/connection_length_histogram.yaml

@@ -11,22 +11,24 @@ tags:
   - elephant
   - rate
 extract:
+  type: aggregates
   aggregates:
-    - name: connection_bytes_hist
-      groupByKeys:
-        - _RecordType
-      operationType: raw_values
-      operationKey: bytes_total
-    - name: connection_bytes_hist_AB
-      groupByKeys:
-        - _RecordType
-      operationType: raw_values
-      operationKey: bytes_AB
-    - name: connection_bytes_hist_BA
-      groupByKeys:
-        - _RecordType
-      operationType: raw_values
-      operationKey: bytes_BA
+    rules:
+      - name: connection_bytes_hist
+        groupByKeys:
+          - _RecordType
+        operationType: raw_values
+        operationKey: bytes_total
+      - name: connection_bytes_hist_AB
+        groupByKeys:
+          - _RecordType
+        operationType: raw_values
+        operationKey: bytes_AB
+      - name: connection_bytes_hist_BA
+        groupByKeys:
+          - _RecordType
+        operationType: raw_values
+        operationKey: bytes_BA
 encode:
   type: prom
   prom:

network_definitions/connection_rate_per_dest_subnet.yaml

@@ -17,12 +17,13 @@ transform:
 extract:
   type: aggregates
   aggregates:
-    - name: dest_connection_subnet_count
-      groupByKeys:
-        - dstSubnet
-        - _RecordType
-      operationType: count
-      operationKey: isNewFlow
+    rules:
+      - name: dest_connection_subnet_count
+        groupByKeys:
+          - dstSubnet
+          - _RecordType
+        operationType: count
+        operationKey: isNewFlow
 encode:
   type: prom
   prom:

network_definitions/connection_rate_per_src_subnet.yaml

@@ -17,11 +17,12 @@ transform:
 extract:
   type: aggregates
   aggregates:
-    - name: src_connection_count
-      groupByKeys:
-        - srcSubnet
-        - _RecordType
-      operationType: count
+    rules:
+      - name: src_connection_count
+        groupByKeys:
+          - srcSubnet
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/connection_rate_per_tcp_flags.yaml

@@ -11,11 +11,12 @@ tags:
 extract:
   type: aggregates
   aggregates:
-    - name: TCPFlags_count
-      groupByKeys:
-        - TCPFlags
-        - _RecordType
-      operationType: count
+    rules:
+      - name: TCPFlags_count
+        groupByKeys:
+          - TCPFlags
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/connections_per_dst_as.yaml

@@ -12,11 +12,12 @@ tags:
 extract:
   type: aggregates
   aggregates:
-    - name: dst_as_connection_count
-      groupByKeys:
-        - dstAS
-        - _RecordType
-      operationType: count
+    rules:
+      - name: dst_as_connection_count
+        groupByKeys:
+          - dstAS
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/connections_per_src_as.yaml

@@ -12,11 +12,12 @@ tags:
 extract:
   type: aggregates
   aggregates:
-    - name: src_as_connection_count
-      groupByKeys:
-        - srcAS
-        - _RecordType
-      operationType: count
+    rules:
+      - name: src_as_connection_count
+        groupByKeys:
+          - srcAS
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/count_per_src_dest_subnet.yaml

@@ -21,13 +21,15 @@ transform:
       type: add_subnet
       parameters: /24
 extract:
+  type: aggregates
   aggregates:
-    - name: count_source_destination_subnet
-      groupByKeys:
-        - dstSubnet24
-        - srcSubnet24
-        - _RecordType
-      operationType: count
+    rules:
+      - name: count_source_destination_subnet
+        groupByKeys:
+          - dstSubnet24
+          - srcSubnet24
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/egress_bandwidth_per_dest_subnet.yaml

@@ -17,13 +17,15 @@ transform:
       type: add_subnet
       parameters: /16
 extract:
+  type: aggregates
   aggregates:
-    - name: bandwidth_destination_subnet
-      groupByKeys:
-        - dstSubnet
-        - _RecordType
-      operationType: sum
-      operationKey: bytes
+    rules:
+      - name: bandwidth_destination_subnet
+        groupByKeys:
+          - dstSubnet
+          - _RecordType
+        operationType: sum
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/egress_bandwidth_per_namespace.yaml

@@ -16,14 +16,16 @@ transform:
       type: add_kubernetes
       parameters: srcK8S_labels
 extract:
+  type: aggregates
   aggregates:
-    - name: bandwidth_namespace
-      groupByKeys:
-        - srcK8S_Namespace
-        - srcK8S_Type
-        - _RecordType
-      operationType: sum
-      operationKey: bytes
+    rules:
+      - name: bandwidth_namespace
+        groupByKeys:
+          - srcK8S_Namespace
+          - srcK8S_Type
+          - _RecordType
+        operationType: sum
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/flows_length_histogram.yaml

@@ -17,13 +17,15 @@ transform:
       type: add_if
       parameters: ">=0"
 extract:
+  type: aggregates
   aggregates:
-    - name: flows_bytes_hist
-      groupByKeys:
-        - all_Evaluate
-        - _RecordType
-      operationType: raw_values
-      operationKey: bytes
+    rules:
+      - name: flows_bytes_hist
+        groupByKeys:
+          - all_Evaluate
+          - _RecordType
+        operationType: raw_values
+        operationKey: bytes
 encode:
   type: prom
   prom:

network_definitions/geo-location_rate_per_dest.yaml

@@ -18,11 +18,12 @@ transform:
 extract:
   type: aggregates
   aggregates:
-    - name: dest_connection_location_count
-      groupByKeys:
-        - dstLocation_CountryName
-        - _RecordType
-      operationType: count
+    rules:
+      - name: dest_connection_location_count
+        groupByKeys:
+          - dstLocation_CountryName
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

network_definitions/network_services_count.yaml

@@ -19,11 +19,12 @@ transform:
 extract:
   type: aggregates
   aggregates:
-    - name: dest_service_count
-      groupByKeys:
-        - service
-        - _RecordType
-      operationType: count
+    rules:
+      - name: dest_service_count
+        groupByKeys:
+          - service
+          - _RecordType
+        operationType: count
 encode:
   type: prom
   prom:

pkg/api/api.go

@@ -54,18 +54,18 @@ const (
 // Note: items beginning with doc: "## title" are top level items that get divided into sections inside api.md.
 
 type API struct {
-	PromEncode         PromEncode          `yaml:"prom" doc:"## Prometheus encode API\nFollowing is the supported API format for prometheus encode:\n"`
-	KafkaEncode        EncodeKafka         `yaml:"kafka" doc:"## Kafka encode API\nFollowing is the supported API format for kafka encode:\n"`
-	S3Encode           EncodeS3            `yaml:"s3" doc:"## S3 encode API\nFollowing is the supported API format for S3 encode:\n"`
-	IngestCollector    IngestCollector     `yaml:"collector" doc:"## Ingest collector API\nFollowing is the supported API format for the NetFlow / IPFIX collector:\n"`
-	IngestKafka        IngestKafka         `yaml:"kafka" doc:"## Ingest Kafka API\nFollowing is the supported API format for the kafka ingest:\n"`
-	IngestGRPCProto    IngestGRPCProto     `yaml:"grpc" doc:"## Ingest GRPC from Network Observability eBPF Agent\nFollowing is the supported API format for the Network Observability eBPF ingest:\n"`
-	TransformGeneric   TransformGeneric    `yaml:"generic" doc:"## Transform Generic API\nFollowing is the supported API format for generic transformations:\n"`
-	TransformFilter    TransformFilter     `yaml:"filter" doc:"## Transform Filter API\nFollowing is the supported API format for filter transformations:\n"`
-	TransformNetwork   TransformNetwork    `yaml:"network" doc:"## Transform Network API\nFollowing is the supported API format for network transformations:\n"`
-	WriteLoki          WriteLoki           `yaml:"loki" doc:"## Write Loki API\nFollowing is the supported API format for writing to loki:\n"`
-	WriteStdout        WriteStdout         `yaml:"stdout" doc:"## Write Standard Output\nFollowing is the supported API format for writing to standard output:\n"`
-	ExtractAggregate   AggregateDefinition `yaml:"aggregates" doc:"## Aggregate metrics API\nFollowing is the supported API format for specifying metrics aggregations:\n"`
-	ConnectionTracking ConnTrack           `yaml:"conntrack" doc:"## Connection tracking API\nFollowing is the supported API format for specifying connection tracking:\n"`
-	ExtractTimebased   ExtractTimebased    `yaml:"timebased" doc:"## Time-based Filters API\nFollowing is the supported API format for specifying metrics time-based filters:\n"`
+	PromEncode         PromEncode       `yaml:"prom" doc:"## Prometheus encode API\nFollowing is the supported API format for prometheus encode:\n"`
+	KafkaEncode        EncodeKafka      `yaml:"kafka" doc:"## Kafka encode API\nFollowing is the supported API format for kafka encode:\n"`
+	S3Encode           EncodeS3         `yaml:"s3" doc:"## S3 encode API\nFollowing is the supported API format for S3 encode:\n"`
+	IngestCollector    IngestCollector  `yaml:"collector" doc:"## Ingest collector API\nFollowing is the supported API format for the NetFlow / IPFIX collector:\n"`
+	IngestKafka        IngestKafka      `yaml:"kafka" doc:"## Ingest Kafka API\nFollowing is the supported API format for the kafka ingest:\n"`
+	IngestGRPCProto    IngestGRPCProto  `yaml:"grpc" doc:"## Ingest GRPC from Network Observability eBPF Agent\nFollowing is the supported API format for the Network Observability eBPF ingest:\n"`
+	TransformGeneric   TransformGeneric `yaml:"generic" doc:"## Transform Generic API\nFollowing is the supported API format for generic transformations:\n"`
+	TransformFilter    TransformFilter  `yaml:"filter" doc:"## Transform Filter API\nFollowing is the supported API format for filter transformations:\n"`
+	TransformNetwork   TransformNetwork `yaml:"network" doc:"## Transform Network API\nFollowing is the supported API format for network transformations:\n"`
+	WriteLoki          WriteLoki        `yaml:"loki" doc:"## Write Loki API\nFollowing is the supported API format for writing to loki:\n"`
+	WriteStdout        WriteStdout      `yaml:"stdout" doc:"## Write Standard Output\nFollowing is the supported API format for writing to standard output:\n"`
+	ExtractAggregate   Aggregates       `yaml:"aggregates" doc:"## Aggregate metrics API\nFollowing is the supported API format for specifying metrics aggregations:\n"`
+	ConnectionTracking ConnTrack        `yaml:"conntrack" doc:"## Connection tracking API\nFollowing is the supported API format for specifying connection tracking:\n"`
+	ExtractTimebased   ExtractTimebased `yaml:"timebased" doc:"## Time-based Filters API\nFollowing is the supported API format for specifying metrics time-based filters:\n"`
 }

pkg/api/extract_aggregate.go

@@ -17,12 +17,19 @@
 
 package api
 
+type Aggregates struct {
+	DefaultExpiryTime Duration             `yaml:"defaultExpiryTime,omitempty" json:"defaultExpiryTime,omitempty" doc:"default time duration of data aggregation to perform rules (default: 2 minutes)"`
+	Rules             AggregateDefinitions `yaml:"rules,omitempty" json:"rules,omitempty" doc:"list of aggregation rules, each includes:"`
+}
+
 type AggregateBy []string
 type AggregateOperation string
+type AggregateDefinitions []AggregateDefinition
 
 type AggregateDefinition struct {
 	Name          string             `yaml:"name,omitempty" json:"name,omitempty" doc:"description of aggregation result"`
 	GroupByKeys   AggregateBy        `yaml:"groupByKeys,omitempty" json:"groupByKeys,omitempty" doc:"list of fields on which to aggregate"`
 	OperationType AggregateOperation `yaml:"operationType,omitempty" json:"operationType,omitempty" doc:"sum, min, max, count, avg or raw_values"`
 	OperationKey  string             `yaml:"operationKey,omitempty" json:"operationKey,omitempty" doc:"internal field on which to perform the operation"`
+	ExpiryTime    Duration           `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"time interval over which to perform the operation"`
 }