This PoC creates a new "SimpleProm" Encode stage that directly extracts metrics from flows without an intermediate Aggregate stage

The rationale is that Prometheus already manages labels aggregations. It leaves more responsibilities (and more power) to the query side. For instance, aggregations such as sum/avg etc. would be performed in PromQL at query time rather than upstream.

Pipelines to generate are simpler as they don't need an "Aggregate" stage, and the prom-encode stage itself has less parameters, e.g.:

	enrichedStage.SimpleEncodePrometheus("prometheus", api.SimplePromEncode{
		Port:   int(b.desired.PrometheusPort),
		Prefix: "netobserv_",
		Metrics: []api.SimplePromMetricsItem{{
			Name:      "bytes_total",
			RecordKey: "Bytes",
			Type:      "counter",
			Labels:    []string{"SrcK8S_Namespace", "DstK8S_Namespace"},
		}, {
			Name:      "packets_total",
			RecordKey: "Packets",
			Type:      "counter",
			Labels:    []string{"SrcK8S_Namespace", "DstK8S_Namespace"},
		}},
	})

I had to create a "fake" stage DecodeCast, just to fix incompatible types map[string]interface{} and GenericMap (which are actually the same). However I think we could do better by merging all Ingest and Decode stages (reach out on slack to discuss)

Took some inspiration from the go-pipes. Allow the same ingest to be directed to multiple pipelines. We allow a single ingest and a single decode stage. After that, we allow any order of transform, extract, encode, write stages. Sample syntax:

pipeline:
  - stage: ingest
    name: ingest1
    params:
      ingest:
        type: file_loop
        file:
          filename: playground/goflow2_input.txt
  - stage: decode
    name: decode1
    follows: ingest1
    params:
      decode:
        type: json
  - stage: transform
    name: generic1
    follows: decode1
    params:
      transform:
        type: generic
        generic:
          rules:
             ..........
  - stage: write
    name: write1
    follows: generic1
    params:
      write:
        type: stdout
  - stage: transform
    name: generic2
    follows: decode1
    params:
      transform:
        type: generic
        generic:
          rules:
             .......
  - stage: write
    name: write2
    follows: generic2
    params:
      write:
        type: stdout

This example shows 2 generic transforms receiving the same flows from the decoder and performing different transforms and different write with them.

The api is not fully consistent, some names are camelCase, other lowcase and sometimes snake_case, e.g:

• valuekey
• expirytime
• readTimeout
• batchSize
• buffer_length

We should agree on a convention to follow

Except for netflow/ipfix which continue to use goflow2 metrics, other ingester now adopt the same "framework" for metrics as the other stages, hence also have new names with a prefix

New metrics are:

ingest_batch_size_bytes

| Name | ingest_batch_size_bytes | |:---|:---| | Description | Ingested batch size distribution, in bytes | | Type | summary | | Labels | stage |

ingest_errors

| Name | ingest_errors | |:---|:---| | Description | Counter of errors during ingestion | | Type | counter | | Labels | stage, type, code |

ingest_flows_processed

| Name | ingest_flows_processed | |:---|:---| | Description | Provides number of flows processed, batches processed, and batch size stats (in number of flows) | | Type | summary | | Labels | stage |

ingest_latency_ms

| Name | ingest_latency_ms | |:---|:---| | Description | Latency between flow end time and ingest time, in milliseconds | | Type | histogram | | Labels | stage |

Fixes #200

@eranra @KalmanMeth Is it ok with you to have a breaking changes here? (people who used legacy v5 now have to configure it explicitly) If that's a problem, we could default to the previous behaviour (legacy port = main port +1) and have another setting to disable v5, but I think it's more simple to keep default as 0 = legacy disabled.

There are many linting errors that stop the project from being built/used.

make build
cmd/confgenerator/main.go:27:2: "github.com/json-iterator/go" imported but not used as jsoniter (typecheck)
        jsoniter "github.com/json-iterator/go"
        ^
cmd/flowlogs-pipeline/main.go:28:2: "github.com/json-iterator/go" imported but not used as jsoniter (typecheck)
        jsoniter "github.com/json-iterator/go"
        ^
pkg/confgen/confgen.go:150:8: undeclared name: `yaml` (typecheck)
        err = yaml.Unmarshal(yamlFile, &defFile)
              ^
pkg/confgen/config.go:78:8: undeclared name: `yaml` (typecheck)
        err = yaml.Unmarshal(yamlFile, &config)
              ^
pkg/confgen/flowlogs2metrics_config.go:115:21: undeclared name: `yaml` (typecheck)
        configData, err := yaml.Marshal(&config)
                           ^
pkg/confgen/confgen.go:29:2: "gopkg.in/yaml.v2" imported but not used (typecheck)
        "gopkg.in/yaml.v2"
        ^
pkg/confgen/config.go:25:2: "gopkg.in/yaml.v2" imported but not used (typecheck)
        "gopkg.in/yaml.v2"
        ^
pkg/confgen/flowlogs2metrics_config.go:24:2: "gopkg.in/yaml.v2" imported but not used (typecheck)
        "gopkg.in/yaml.v2"
        ^
pkg/pipeline/pipeline_test.go:25:2: "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/decode" imported but not used (typecheck)
        "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/decode"
        ^
pkg/pipeline/pipeline_test.go:26:2: "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest" imported but not used (typecheck)
        "github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest"
        ^
pkg/pipeline/encode/encode_prom.go:21:2: could not import container/list (-: could not load export data: cannot import "container/list" (unstable iexport format version 2, just rebuild compiler and std library), export data is newer version - update tool) (typecheck)
        "container/list"
        ^
pkg/pipeline/transform/transform_network.go:54:25: n.Rules undefined (type *Network has no field or method Rules) (typecheck)
        for _, rule := range n.Rules {
                               ^
pkg/pipeline/transform/connection_tracking/connection_tracking.go:21:2: could not import container/list (-: could not load export data: cannot import "container/list" (unstable iexport format version 2, just rebuild compiler and std library), export data is newer version - update tool) (typecheck)
        "container/list"
        ^
pkg/pipeline/transform/kubernetes/kubernetes_test.go:86:15: informerMock.On undefined (type *InformerMock has no field or method On) (typecheck)
        informerMock.On("GetIndexer", mock.Anything, mock.Anything, mock.Anything).Return(nil)
                     ^
pkg/pipeline/transform/kubernetes/kubernetes.go:28:2: "k8s.io/api/apps/v1" imported but not used as appsv1 (typecheck)
        appsv1 "k8s.io/api/apps/v1"
        ^
pkg/pipeline/transform/location/location.go:54:17: undeclared name: `ip2location` (typecheck)
var locationDB *ip2location.DB
                ^
pkg/pipeline/transform/location/location.go:146:19: undeclared name: `ip2location` (typecheck)
        db, openDBErr := ip2location.OpenDB(DBFileLocation + "/" + DBFilename)
                         ^
pkg/pipeline/transform/location/location_test.go:125:16: undeclared name: `ip2location` (typecheck)
        locationDB = &ip2location.DB{}
                      ^
pkg/pipeline/transform/location/location.go:33:2: "github.com/ip2location/ip2location-go/v9" imported but not used (typecheck)
        "github.com/ip2location/ip2location-go/v9"
        ^
pkg/pipeline/transform/location/location_test.go:31:2: "github.com/ip2location/ip2location-go/v9" imported but not used (typecheck)
        "github.com/ip2location/ip2location-go/v9"
        ^
pkg/pipeline/write/write_loki_test.go:108:5: fe.On undefined (type fakeEmitter has no field or method On) (typecheck)
        fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
           ^
pkg/pipeline/write/write_loki_test.go:118:5: fe.AssertCalled undefined (type fakeEmitter has no field or method AssertCalled) (typecheck)
        fe.AssertCalled(t, "Handle", model.LabelSet{
           ^
pkg/pipeline/write/write_loki_test.go:124:5: fe.AssertCalled undefined (type fakeEmitter has no field or method AssertCalled) (typecheck)
        fe.AssertCalled(t, "Handle", model.LabelSet{
           ^
pkg/pipeline/write/write_loki_test.go:161:7: fe.On undefined (type fakeEmitter has no field or method On) (typecheck)
                        fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
                           ^
pkg/pipeline/write/write_loki_test.go:165:7: fe.AssertCalled undefined (type fakeEmitter has no field or method AssertCalled) (typecheck)
                        fe.AssertCalled(t, "Handle", model.LabelSet{},
                           ^
pkg/pipeline/write/write_loki_test.go:204:7: fe.On undefined (type fakeEmitter has no field or method On) (typecheck)
                        fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
                           ^
make: *** [Makefile:69: lint] Error 1

golangci-lint version
golangci-lint has version 1.39.0 built from 9aea4aee on 2021-03-26T08:02:53Z

Align the network transform functionality of flowlogs2metrics with https://github.com/netobserv/goflow2-kube-enricher so that the information and fields added by the network transform phase are equivalent. This will allow smooth integration with NOO.

This PR removes the batching logic in the Kafka ingester and processes flows 1 by 1 on each pipeline stage.

While the CPU usage doesn't seem to noticeably decrease:

It greatly improves memory consumption:

(as well as simplifies some parts of the code)

This PR adds a Transformer for each Kubernetes informer that converts the watched Pods, Services, and Nodes to Info instances, that are stored by the informers cache and contain only the data that is required for the decoration.

This also alleviates the load on the Garbage Collector: before this patch, a new Info instance was created for each flow decoration. Now the same objects are reused on GetInfo invocations for the same IP.

In Flowlogs-Pipelines with mid-to-low loads, no improvements have been observed in terms of CPU and slight improvements in terms of memory. In the image below, the middle part corresponds to the new version of FLP.

If the profile.port argument is set to something different than 0, it will add a pprof endpoint so we can inspect what's going on FLP in terms of memory and CPU.

JIRA: https://issues.redhat.com/browse/NETOBSERV-386

Allows to configure IP ranges and assign them any name IPs falling in those ranges are flagged with that name

This update the Dockerfile to make it possible to cross build FLP container image.

Example of manual cross build command: podman buildx build --platform=linux/ppc64le -t test -f ./contrib/docker/Dockerfile .

Could someone with docker installed could confirm that the cli API is identical and the example above also works with docker?

This PR allows configuring different timeouts for connections based on their key fields.

conntrack:
  keyDefinition:
    fieldGroups:
    - name: protocol
      fields:
      - Proto
    # ...
  scheduling:
  - selector: # UDP connections
      Proto: 17
    endConnectionTimeout: 5s
    updateConnectionInterval: 40s
  - selector: {} # Default group
    endConnectionTimeout: 10s
    updateConnectionInterval: 30s

Implementation: Instead of using a single Multi-Order Map in the connection store for all the connections, the connection store will have a Multi-Order Map for each scheduling group. In addition, the store has a new hashId2groupIdx map that given a connection hash ID, finds the Multi-Order Map of the connection.

TODO:

• [ ] Add a scheduling group label to the conntrack_memory_connections metric. This will allow monitoring the number of active connections per scheduling group.

Currently, we have only a single hash ID for a connection which is a hash on some selected fields. The flow direction (i.e. "ingress" or "egress") is also added to these fields to eliminate aggregation of duplicate stats (such as bytes and packets). This makes 2 different hashes for the same connection. The problem is that in the console plugin we would like to group together the records of the 2 hashes. One option to do so is to add another hash ID that excludes the flow direction field. This will generate the same hash regardless of the flow direction. This new hash can be used in the console plugin to group records by.

This PR adds IPFIX export support for FLP o/p:

Way to enable this in config, for e.g.:

    write:
      type: ipfix
      ipfix:
        targetHost: 127.0.0.1
        targetPort: 9998
        transport: tcp
        enterpriseID: 27207

enterpriseID is a custom registry ID for kubernetes transformations, and other non-standard entries.

Main Changes in the following files:

1. pkg/api/write_ipfix.go
2. pkg/api/api.go
3. pkg/config/config.go
4. pkg/pipeline/pipeline_builder.go
5. pkg/pipeline/write/write_ipfix.go
6. go.mod