New component: Community ID Network Flow Hash Processor

[mashhurs]

The purpose and use-cases of the new component

(Please see the discussion for details and decided to make it with processor: #34062)

When monitoring/analyzing network flow, for example threat hunting security use cases, it's often required to make "joins" on network source and destination info. A community ID is a concept to aggregate/simplifie joins (extra effort) to provide a better user experience when analyzing the data (aggregate by community ID to collect statistics, etc.)

The Community ID is a single unique ID based on the network flow info. It is an additional flow identifier and doesn't replace existing flow identification mechanisms already supported by the monitors. See the specification.

At Elastic (it is not a vendor specific), the best practise examples:

• Elastic beats: https://www.elastic.co/guide/en/beats/filebeat/current/community-id.html
• Elasticsearch community_id: https://www.elastic.co/guide/en/beats/filebeat/current/community-id.html

Initial implementation

Initial implementation will be minimum configurable embracing the KISS.

Mapping fields:

• [IN] source ip: source.address
• [IN]source port: source.port
• [IN]destination ip: destination.address
• [IN]destination port: destination.port
• [IN]transport protocol: network.transport
• [OUT] target: defaults to network.community_id - discussion/investigation required
  • (todo-1) double check naming convention
  • (todo-2) add sub-field to network field by default
  • (todo-3) best practise when adding fields

Validation:

• Parse IP against IPv4|IPv6, error if invalid IP
• If network.type is available, error if IP mismatches
• Error if port is not in range of 1 ~ 65535
• Error if transport protocol is not TCP|UDP|SCTP

Future enhancement

• configurable source and destination attributes which useful when using after DNS resolver processor
• IANA number: need a field specification
• ICMP type: to support ICMP, need a field specification
• ICMP code: to support ICMP, need a field specification

Example configuration for the component

processors:
  communityid:
    # super simple stupid config, by default uses `source.address`, `source.port`, `destination.address`, `destination.port` and `network.transport` to calculate the flow

    # a target field to be assigned with calculated community ID, defaults to `network.community_id`
    target: network.community_id

    # a seed value to prevent hash collisions between network domains, such as a staging and production network that use the same addressing scheme. Defaults to 0 (zero).
    seed: 0

    # Defines how the processor handles errors. Accepts `log` or `silent`
    # - `log`: Processor logs errors and continues to the next processor (default).
    # - `silent`: Processor ignores errors without logging and continues to the next processor.
    error_mode: log

Example in/out

// input
{
  "resourceMetrics": [
    {
      "resource": {
        "attributes": [
          {
            "key": "service.name",
            "value": { "stringValue": "my.service" }
          },
          {
            "key": "timestamp",
            "value": { "stringValue": "2025-05-21T09:25:18Z" }
          },
          {
            "key": "source.ip",
            "value": { "stringValue": "123.124.125.126" }
          },
          {
            "key": "source.port",
            "value": { "intValue": "12345" }
          },
          {
            "key": "destination.ip",
            "value": { "stringValue": "55.56.57.58" }
          },
          {
            "key": "destination.port",
            "value": { "intValue": "80" }
          }
        ]
      },
      "scopeMetrics": [
        {
          "scope": {
            "name": "my.library",
            "version": "1.0.0",
            "attributes": [
              {
                "key": "my.scope.attribute",
                "value": {
                  "stringValue": "some scope attribute"
                }
              }
            ]
          },
          "metrics": [
            ...
          ]
        }
      ]
    }
  ]
}

// output in Log exporter
2025-05-21T12:00:00Z    DEBUG    loggingexporter/logging_exporter.go:someline
ResourceMetrics #0
Resource labels:
     -> service.name: STRING(my.service)
     -> timestamp: STRING(2025-05-21T09:25:18Z)
     -> source.ip: STRING(123.124.125.126)
     -> source.port: INT(12345)
     -> destination.ip: STRING(55.56.57.58)
     -> destination.port: INT(80)
     -> network.community_id: STRING(1:9qr9Z1LViXcNwtLVOHZ3CL8MlyM=)
ScopeMetrics #0
Scope name: example

Telemetry data types supported

traces, metrics and logs

Is this a vendor-specific component?

• This is a vendor-specific component
• If this is a vendor-specific component, I am a member of the OpenTelemetry organization.
• If this is a vendor-specific component, I am proposing to contribute and support it as a representative of the vendor.

Not a vendor specific.

• early adoption of vendors/solutions such as Suricate, Zeek, MISP, etc..: https://zeek.org/2019/07/31/an-update-on-community-id/
• Wireshark: https://corelight.com/blog/community-id-support-for-wireshark
• Tenzip, a security data lake: Implement generic Community ID computation API tenzir/tenzir#525
• List of vendor implementations: https://github.com/corelight/community-id-spec?tab=readme-ov-file#production-implementations

Code Owner(s)

evan-bradley

Sponsor (optional)

No response

Additional context

No response

[kentquirk]

The description above tells how a community ID is calculated, but it doesn't explain how the payload is transformed. Please add that information to the description.

[atoulme]

Is @evan-bradley quoted here as a potential sponsor?

[mashhurs]

Is @evan-bradley quoted here as a potential sponsor?

Initially I had a proposal to make this functionality with ottl. After discussing (here) with @evan-bradley, we decided to move to processor.

[mashhurs]

The description above tells how a community ID is calculated, but it doesn't explain how the payload is transformed. Please add that information to the description.

The way I was thinking is, the flow calculation will be based on source and destination attributes. The result will be included network (configurable) attribute. Sample configuration is attached in the description but I don't know yet the implementation details (since I didn't actively start the implementation). My knowledge/experience on oTel is so limited, so please let me know if you need more info or it needs to be designed in a specific way or your thoughts for this feature/implementation.
Thanks.

[github-actions]

This issue has been inactive for 60 days. It will be closed in 60 days if there is no activity. To ping code owners by adding a component label, see Adding Labels via Comments, or if you are unsure of which component this issue relates to, please ping @open-telemetry/collector-contrib-triagers. If this issue is still relevant, please ping the code owners or leave a comment explaining why it is still relevant. Otherwise, please close it.

[github-actions]

This issue has been closed as inactive because it has been stale for 120 days with no activity.