Community ID processor

The community_id processor is used to generate the community ID flow hash for network flow tuples. The community ID flow hash algorithm is defined in the community ID specification. The processor-generated hash value can be used to correlate all related network events so that you can filter the network flow data by the hash value or generate statistics by aggregating on the hash field. The processor supports the TCP, UDP, SCTP, ICMP, and IPv6-ICMP network protocols. The SHA-1 hash algorithm is used to generate the hash value.

The following is the community_id processor syntax:

  1. {
  2.   "community_id": {
  3.     "source_ip_field": "source_ip",
  4.     "source_port_field": "source_port",
  5.     "destination_ip_field": "destination_ip",
  6.     "destination_port_field": "destination_port",
  7.     "iana_protocol_number_field": "iana_protocol_number",
  8.     "source_port_field": "source_port",
  9.     "target_field": "community_id"
  10.   }
  11. }

copy

Configuration parameters

The following table lists the required and optional parameters for the community_id processor.

ParameterRequired/OptionalDescription
source_ip_fieldRequiredThe name of the field containing the source IP address.
source_port_fieldOptionalThe name of the field containing the source port address. If the network protocol is TCP, UDP, or SCTP, then the field is required. Otherwise, it is not required.
destination_ip_fieldRequiredThe name of the field containing the destination IP address.
destination_port_fieldOptionalThe name of the field containing the destination port address. If the network protocol is TCP, UDP, or SCTP, then the field is required. Otherwise, it is not required.
iana_protocol_numberOptionalThe name of the field containing the protocol number defined by the Internet Assigned Numbers Authority (IANA). The supported values are 1 (ICMP), 6 (TCP), 17 (UDP), 58 (IPv6-ICMP), and 132 (SCTP).
protocol_fieldOptionalThe name of the field containing the protocol name. If iana_protocol_number is not set, then the field is required. Otherwise, it is not required.
icmp_type_fieldOptionalThe name of the field containing the ICMP message type. Required when the protocol is ICMP or IPv6-ICMP.
icmp_code_fieldOptionalThe name of the field containing the ICMP message code. For certain ICMP message types that do not have a code, the field is optional. Otherwise, it is required.
seedOptionalThe seed for generating the community ID hash. The value must be between 0 and 65535.
target_fieldOptionalThe name of the field in which to store the community ID hash value. Default target field is community_id.
ignore_missingOptionalSpecifies whether the processor should exit quietly if one of the required fields is missing. Default is false.
descriptionOptionalA brief description of the processor.
ifOptionalA condition for running the processor.
ignore_failureOptionalIf set to true, then failures are ignored. Default is false.
on_failureOptionalA list of processors to run if the processor fails.
tagOptionalAn identifier tag for the processor. Useful for debugging in order to distinguish between processors of the same type.

Using the processor

Follow these steps to use the processor in a pipeline.

Step 1: Create a pipeline

The following query creates a pipeline named community_id_pipeline that uses the community_id processor to generate a hash value for the network flow tuple:

  1. PUT /_ingest/pipeline/commnity_id_pipeline
  2. {
  3.   "description": "generate hash value for the network flow tuple",
  4.   "processors": [
  5.     {
  6.       "community_id": {
  7.         "source_ip_field": "source_ip",
  8.         "source_port_field": "source_port",
  9.         "destination_ip_field": "destination_ip",
  10.         "destination_port_field": "destination_port",
  11.         "iana_protocol_number_field": "iana_protocol_number",
  12.         "target_field": "community_id"
  13.      }
  14.     }
  15.   ]
  16. }

copy

Step 2 (Optional): Test the pipeline

It is recommended that you test your pipeline before ingesting documents.

To test the pipeline, run the following query:

  1. POST _ingest/pipeline/commnity_id_pipeline/_simulate
  2. {
  3.   "docs": [
  4.     {
  5.       "_index": "testindex1",
  6.       "_id": "1",
  7.       "_source": {
  8.         "source_ip": "66.35.250.204",
  9.         "source_port": 80,
  10.         "destination_ip": "128.232.110.120",
  11.         "destination_port": 34855,
  12.         "iana_protocol_number": 6
  13.       }
  14.     }
  15.   ]
  16. }

copy

Response

The following example response confirms that the pipeline is working as expected:

  1. {
  2.   "docs": [
  3.     {
  4.       "doc": {
  5.         "_index": "testindex1",
  6.         "_id": "1",
  7.         "_source": {
  8.           "community_id": "1:LQU9qZlK+B5F3KDmev6m5PMibrg=",
  9.           "destination_ip": "128.232.110.120",
  10.           "destination_port": 34855,
  11.           "source_port": 80,
  12.           "iana_protocol_number": 6,
  13.           "source_ip": "66.35.250.204"
  14.         },
  15.         "_ingest": {
  16.           "timestamp": "2024-03-11T02:17:22.329823Z"
  17.         }
  18.       }
  19.     }
  20.   ]
  21. }

Step 3: Ingest a document

The following query ingests a document into an index named testindex1:

  1. PUT testindex1/_doc/1?pipeline=commnity_id_pipeline
  2. {
  3.   "source_ip": "66.35.250.204",
  4.   "source_port": 80,
  5.   "destination_ip": "128.232.110.120",
  6.   "destination_port": 34855,
  7.   "iana_protocol_number": 6
  8. }

copy

Response

The request indexes the document into the testindex1 index:

  1. {
  2.   "_index": "testindex1",
  3.   "_id": "1",
  4.   "_version": 1,
  5.   "result": "created",
  6.   "_shards": {
  7.     "total": 2,
  8.     "successful": 1,
  9.     "failed": 0
  10.   },
  11.   "_seq_no": 0,
  12.   "_primary_term": 1
  13. }

Step 4 (Optional): Retrieve the document

To retrieve the document, run the following query:

  1. GET testindex1/_doc/1

copy