Merge pull request #225 from jotak/no-decode-stage

Remove Decoder stage (breaking change)

Author: jotak

README.md

@@ -112,14 +112,12 @@ FLP is a framework. The main FLP object is the **pipeline**. FLP **pipeline** ca
 
 The pipeline is constructed of a sequence of stages. Each stage is classified into one of the following types:
 - **ingest** - obtain flows from some source, one entry per line
-- **decode** - parse input lines into a known format, e.g., dictionary (map) of AWS or goflow data
 - **transform** - convert entries into a standard format; can include multiple transform stages
 - **write** - provide the means to write the data to some target, e.g. loki, standard output, etc
 - **extract** - derive a set of metrics from the imported flows
 - **encode** - make the data available in appropriate format (e.g. prometheus)
 
 The first stage in a pipeline must be an **ingest** stage.
-The **ingest** stage is typically followed by a **decode** stage, unless the **ingest** stage also performs the decoding.
 Each stage (other than an **ingest** stage) specifies the stage it follows.
 Multiple stages may follow from a particular stage, thus allowing the same data to be consumed by multiple parallel pipelines.
 For example, multiple **transform** stages may be performed and the results may be output to different targets.
@@ -132,14 +130,12 @@ A full configuration file with the data consumed by  two different transforms mi
 ```yaml
 pipeline:
   - name: ingest1
-  - name: decode1
-    follows: ingest1
   - name: generic1
-    follows: decode1
+    follows: ingest1
   - name: write1
     follows: generic1
   - name: generic2
-    follows: decode1
+    follows: ingest1
   - name: write2
     follows: generic2
 parameters:
@@ -148,9 +144,8 @@ parameters:
       type: file_loop
       file:
         filename: hack/examples/ocp-ipfix-flowlogs.json
-  - name: decode1
-    decode:
-      type: json
+        decoder:
+          type: json
   - name: generic1
     transform:
       type: generic
@@ -200,19 +195,16 @@ For example:
 log-level: info
 pipeline:
   - name: ingest_file
-  - name: decode_json
-    follows: ingest_file
   - name: write_stdout
-    follows: write_stdout
-parameters
-  - name ingest_file
+    follows: ingest_file
+parameters:
+  - name: ingest_file
     ingest:
       type: file
       file:
         filename: hack/examples/ocp-ipfix-flowlogs.json
-  - name: decode_json
-    decode:
-      type: json
+        decoder:
+          type: json
   - name: write_stdout
     write:
       type: stdout
@@ -224,11 +216,11 @@ parameters
 
 2. Using command line parameters:
 
-`./flowlogs-pipeline --pipeline "[{\"name\":\"ingest1\"},{\"follows\":\"ingest1\",\"name\":\"decode1\"},{\"follows\":\"decode1\",\"name\":\"write1\"}]" --parameters "[{\"ingest\":{\"file\":{\"filename\":\"hack/examples/ocp-ipfix-flowlogs.json\"},\"type\":\"file\"},\"name\":\"ingest1\"},{\"decode\":{\"type\":\"json\"},\"name\":\"decode1\"},{\"name\":\"write1\",\"write\":{\"type\":\"stdout\"}}]"`
+`./flowlogs-pipeline --pipeline "[{\"name\":\"ingest1\"},{\"follows\":\"ingest1\",\"name\":\"write1\"}]" --parameters "[{\"ingest\":{\"file\":{\"filename\":\"hack/examples/ocp-ipfix-flowlogs.json\"},\"decoder\":{\"type\":\"json\"},\"type\":\"file\"},\"name\":\"ingest1\"},{\"name\":\"write1\",\"write\":{\"type\":\"stdout\"}}]"`
 
 Options included in the command line override the options specified in the config file.
 
-`flowlogs-pipeline --log-level debug --pipeline "[{\"name\":\"ingest1\"},{\"follows\":\"ingest1\",\"name\":\"decode1\"},{\"follows\":\"decode1\",\"name\":\"write1\"}]" --config <configFile>`
+`flowlogs-pipeline --log-level debug --pipeline "[{\"name\":\"ingest1\"},{\"follows\":\"ingest1\",\"name\":\"write1\"}]" --config <configFile>`
 
 3. TODO: environment variables
 

contrib/benchmarks/baseline/config.yaml

@@ -4,8 +4,8 @@ pipeline:
     type: file
     file:
       filename: ../../contrib/benchmarks/baseline/log-lines.json
-  decode:
-    type: json
+      decoder:
+        type: json
   encode:
     type: none
   extract:

contrib/benchmarks/transform/config.yaml

@@ -4,8 +4,8 @@ pipeline:
     type: file
     file:
       filename: ../../contrib/benchmarks/baseline/log-lines.json
-  decode:
-    type: json
+      decoder:
+        type: json
   encode:
     type: none
   extract:

contrib/kubernetes/flowlogs-pipeline.conf.yaml

@@ -8,9 +8,6 @@ parameters:
       portLegacy: 2056
     type: collector
   name: ingest_collector
-- decode:
-    type: json
-  name: decode_json
 - name: transform_generic
   transform:
     generic:
@@ -328,8 +325,6 @@ parameters:
 pipeline:
 - name: ingest_collector
 - follows: ingest_collector
-  name: decode_json
-- follows: decode_json
   name: transform_generic
 - follows: transform_generic
   name: transform_network

docs/api.md

@@ -59,6 +59,10 @@ Following is the supported API format for the kafka ingest:
          groupBalancers: list of balancing strategies (range, roundRobin, rackAffinity)
          startOffset: FirstOffset (least recent - default) or LastOffset (most recent) offset available for a partition
          batchReadTimeout: how often (in milliseconds) to process input
+         decoder: decoder to use (E.g. json or protobuf)
+             type: (enum) one of the following:
+                 json: JSON decoder
+                 protobuf: Protobuf decoder
 </pre>
 ## Ingest GRPC from Network Observability eBPF Agent
 Following is the supported API format for the Network Observability eBPF ingest:
@@ -68,13 +72,6 @@ Following is the supported API format for the Network Observability eBPF ingest:
          port: the port number to listen on
          bufferLength: the length of the ingest channel buffer, in groups of flows, containing each group hundreds of flows (default: 100)
 </pre>
-## Aws ingest API
-Following is the supported API format for Aws flow entries:
-
-<pre>
- aws:
-         fields: list of aws flow log fields
-</pre>
 ## Transform Generic API
 Following is the supported API format for generic transformations:
 

hack/deploy-and-monitor-k8s-network-workload.sh

@@ -31,8 +31,6 @@ pipeline:
       port: 2055
       portLegacy: 2056
     type: collector
-  decode:
-    type: json
   encode:
     type: none
   extract:

hack/examples/ocp-ipfix-config.yaml

@@ -3,7 +3,7 @@ pipeline:
     type: file
     file:
       filename: hack/examples/ocp-ipfix-flowlogs.json
-  decode:
-    type: json
+      decoder:
+        type: json
   transform:
     type: generic

pkg/api/api.go

@@ -24,9 +24,6 @@ const (
 	CollectorType         = "collector"
 	GRPCType              = "grpc"
 	KafkaType             = "kafka"
-	JSONType              = "json"
-	PBType                = "protobuf"
-	AWSType               = "aws"
 	StdoutType            = "stdout"
 	LokiType              = "loki"
 	AggregateType         = "aggregates"
@@ -56,7 +53,6 @@ type API struct {
 	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"`
-	DecodeAws        DecodeAws           `yaml:"aws" doc:"## Aws ingest API\nFollowing is the supported API format for Aws flow entries:\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"`

pkg/api/decode_aws.go

@@ -0,0 +1,14 @@
+package api
+
+type Decoder struct {
+	Type string `yaml:"type" json:"type" enum:"DecoderEnum" doc:"one of the following:"`
+}
+
+type DecoderEnum struct {
+	JSON     string `yaml:"json" json:"json" doc:"JSON decoder"`
+	Protobuf string `yaml:"protobuf" json:"protobuf" doc:"Protobuf decoder"`
+}
+
+func DecoderName(decoder string) string {
+	return GetEnumName(DecoderEnum{}, decoder)
+}