[Kafka] Support Producer and Consumer datastreams in the Kafka Integration

packages/kafka/data_stream/consumer/agent/stream/stream.yml.hbs

@@ -0,0 +1,28 @@
+metricsets: ["consumer"]
+period: {{period}}
+hosts:
+{{#each jolokia_hosts}}
+  - {{this}}
+{{/each}}
+{{#if ssl.certificate}}
+ssl.certificate: {{ssl.certificate}}
+{{/if}}
+{{#if ssl.certificate_authorities}}
+ssl.certificate_authorities: {{ssl.certificate_authorities}}
+{{/if}}
+{{#if ssl.key}}
+ssl.key: {{ssl.key}}
+{{/if}}
+{{#if ssl.key_passphrase}}
+ssl.key_passphrase: {{ssl.key_passphrase}}
+{{/if}}
+{{#if ssl.verification_mode}}
+ssl.verification_mode: {{ssl.verification_mode}}
+{{/if}}
+{{#if ssl.ca_trusted_fingerprint}}
+ssl.ca_trusted_fingerprint: {{ssl.ca_trusted_fingerprint}}
+{{/if}}
+{{#if processors}}
+processors:
+{{processors}}
+{{/if}}

packages/kafka/data_stream/consumer/fields/agent.yml

@@ -0,0 +1,94 @@
+- name: cloud
+  title: Cloud
+  group: 2
+  description: Fields related to the cloud or infrastructure the events are coming from.
+  footnote: 'Examples: If Metricbeat is running on an EC2 host and fetches data from its host, the cloud info contains the data about this machine. If Metricbeat runs on a remote machine outside the cloud and fetches data from a service running in the cloud, the field contains cloud data from the machine the service is running on.'
+  type: group
+  fields:
+    - name: account.id
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: 'The cloud account or organization id used to identify different entities in a multi-tenant environment.  Examples: AWS account id, Google Cloud ORG Id, or other unique identifier.'
+      example: 666777888999
+    - name: availability_zone
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: Availability zone in which this host is running.
+      example: us-east-1c
+    - name: instance.id
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      description: Instance ID of the host machine.
+      example: i-1234567890abcdef0
+      dimension: true
+    - name: provider
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      description: Name of the cloud provider. Example values are aws, azure, gcp, or digitalocean.
+      example: aws
+      dimension: true
+    - name: region
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: Region in which this host is running.
+      example: us-east-1
+    - name: image.id
+      type: keyword
+      description: Image ID for the cloud instance.
+- name: container
+  title: Container
+  group: 2
+  description: 'Container fields are used for meta information about the specific container that is the source of information.  These fields help correlate data based containers from any runtime.'
+  type: group
+  fields:
+    - name: id
+      level: core
+      type: keyword
+      ignore_above: 1024
+      description: Unique container id.
+      dimension: true
+- name: host
+  title: Host
+  group: 2
+  description: 'A host is defined as a general computing instance.  ECS host.* fields should be populated with details about the host on which the event happened, or from which the measurement was taken. Host types include hardware, virtual machines, Docker containers, and Kubernetes nodes.'
+  type: group
+  fields:
+    - name: name
+      level: core
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: 'Name of the host.  It can contain what `hostname` returns on Unix systems, the fully qualified domain name, or a name specified by the user. The sender decides which value to use.'
+    - name: containerized
+      type: boolean
+      description: >
+        If the host is a container.
+
+    - name: os.build
+      type: keyword
+      example: "18D109"
+      description: >
+        OS build information.
+
+    - name: os.codename
+      type: keyword
+      example: "stretch"
+      description: >
+        OS codename, if any.
+
+- name: agent
+  title: Agent
+  type: group
+  fields:
+    - name: id
+      type: keyword
+      ignore_above: 1024
+      dimension: true

packages/kafka/data_stream/consumer/fields/base-fields.yml

@@ -0,0 +1,20 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.
+- name: event.module
+  type: constant_keyword
+  description: Event module
+  value: kafka
+- name: event.dataset
+  type: constant_keyword
+  description: Event dataset
+  value: kafka.consumer

packages/kafka/data_stream/consumer/fields/ecs.yml

@@ -0,0 +1,3 @@
+- external: ecs
+  name: service.address
+  dimension: true

packages/kafka/data_stream/consumer/fields/fields.yml

@@ -0,0 +1,31 @@
+- name: consumer
+  type: group
+  description: Consumer metrics from Kafka Consumer JMX
+  fields:
+    - name: mbean
+      description: Mbean that this event is related to
+      type: keyword
+    - name: fetch_rate
+      description: The minimum rate at which the consumer sends fetch requests to a broker
+      type: float
+    - name: bytes_consumed
+      description: The average number of bytes consumed for a specific topic per second
+      type: float
+    - name: records_consumed
+      description: The average number of records consumed per second for a specific topic
+      type: float
+    - name: in.bytes_per_sec
+      description: The rate of bytes coming in to the consumer
+      type: float
+    - name: max_lag
+      description: The maximum consumer lag
+      type: float
+    - name: zookeeper_commits
+      description: The rate of offset commits to ZooKeeper
+      type: float
+    - name: kafka_commits
+      description: The rate of offset commits to Kafka
+      type: float
+    - name: messages_in
+      description: The rate of consumer message consumption
+      type: float

packages/kafka/data_stream/consumer/fields/package-fields.yml

@@ -0,0 +1,40 @@
+- name: kafka
+  type: group
+  fields:
+    - name: broker
+      type: group
+      fields:
+        - name: id
+          type: long
+          description: |
+            Broker id
+        - name: address
+          type: keyword
+          #Reason to add as dimension field: Multiple brokers may exist in a Kafka cluster.
+          dimension: true
+          description: |
+            Broker advertised address
+    - name: topic.name
+      type: keyword
+      #Reason to add as dimension field: Multiple values of topics exist.
+      dimension: true
+      description: |
+        Topic name
+    - name: topic.error.code
+      type: long
+      description: |
+        Topic error code.
+    - name: partition.id
+      type: long
+      description: |
+        Partition id.
+    - name: partition.topic_id
+      type: keyword
+      #Reason to add as dimension field: Multiple records exist for the same kafka.partition.id, kafka.topic.name
+      dimension: true
+      description: Unique id of the partition in the topic.
+    - name: partition.topic_broker_id
+      type: keyword
+      #Reason to add as dimension field: For future use. 
+      dimension: true
+      description: Unique id of the partition in the topic and the broker.

packages/kafka/data_stream/consumer/manifest.yml

@@ -0,0 +1,25 @@
+title: Kafka consumer metrics
+type: metrics
+streams:
+  - input: kafka/metrics
+    title: Kafka consumer metrics
+    description: Collect Kafka consumer metrics
+    vars:
+      - name: jolokia_hosts
+        type: text
+        title: Address of Jolokia agent installed in Kafka
+        multi: true
+        required: true
+        show_user: true
+        default:
+          - 'localhost:8774'
+      - name: processors
+        type: yaml
+        title: Processors
+        multi: false
+        required: false
+        show_user: false
+        description: >
+          Processors are used to reduce the number of fields in the exported event or to enhance the event with metadata. This executes in the agent before the events are shipped. See [Processors](https://www.elastic.co/guide/en/fleet/current/elastic-agent-processor-configuration.html) for details.
+elasticsearch:
+  index_mode: "time_series"

packages/kafka/data_stream/producer/agent/stream/stream.yml.hbs

@@ -0,0 +1,28 @@
+metricsets: ["producer"]
+period: {{period}}
+hosts:
+{{#each jolokia_hosts}}
+  - {{this}}
+{{/each}}
+{{#if ssl.certificate}}
+ssl.certificate: {{ssl.certificate}}
+{{/if}}
+{{#if ssl.certificate_authorities}}
+ssl.certificate_authorities: {{ssl.certificate_authorities}}
+{{/if}}
+{{#if ssl.key}}
+ssl.key: {{ssl.key}}
+{{/if}}
+{{#if ssl.key_passphrase}}
+ssl.key_passphrase: {{ssl.key_passphrase}}
+{{/if}}
+{{#if ssl.verification_mode}}
+ssl.verification_mode: {{ssl.verification_mode}}
+{{/if}}
+{{#if ssl.ca_trusted_fingerprint}}
+ssl.ca_trusted_fingerprint: {{ssl.ca_trusted_fingerprint}}
+{{/if}}
+{{#if processors}}
+processors:
+{{processors}}
+{{/if}}

packages/kafka/data_stream/producer/fields/agent.yml

@@ -0,0 +1,94 @@
+- name: cloud
+  title: Cloud
+  group: 2
+  description: Fields related to the cloud or infrastructure the events are coming from.
+  footnote: 'Examples: If Metricbeat is running on an EC2 host and fetches data from its host, the cloud info contains the data about this machine. If Metricbeat runs on a remote machine outside the cloud and fetches data from a service running in the cloud, the field contains cloud data from the machine the service is running on.'
+  type: group
+  fields:
+    - name: account.id
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: 'The cloud account or organization id used to identify different entities in a multi-tenant environment.  Examples: AWS account id, Google Cloud ORG Id, or other unique identifier.'
+      example: 666777888999
+    - name: availability_zone
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: Availability zone in which this host is running.
+      example: us-east-1c
+    - name: instance.id
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      description: Instance ID of the host machine.
+      example: i-1234567890abcdef0
+      dimension: true
+    - name: provider
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      description: Name of the cloud provider. Example values are aws, azure, gcp, or digitalocean.
+      example: aws
+      dimension: true
+    - name: region
+      level: extended
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: Region in which this host is running.
+      example: us-east-1
+    - name: image.id
+      type: keyword
+      description: Image ID for the cloud instance.
+- name: container
+  title: Container
+  group: 2
+  description: 'Container fields are used for meta information about the specific container that is the source of information.  These fields help correlate data based containers from any runtime.'
+  type: group
+  fields:
+    - name: id
+      level: core
+      type: keyword
+      ignore_above: 1024
+      description: Unique container id.
+      dimension: true
+- name: host
+  title: Host
+  group: 2
+  description: 'A host is defined as a general computing instance.  ECS host.* fields should be populated with details about the host on which the event happened, or from which the measurement was taken. Host types include hardware, virtual machines, Docker containers, and Kubernetes nodes.'
+  type: group
+  fields:
+    - name: name
+      level: core
+      type: keyword
+      ignore_above: 1024
+      dimension: true
+      description: 'Name of the host.  It can contain what `hostname` returns on Unix systems, the fully qualified domain name, or a name specified by the user. The sender decides which value to use.'
+    - name: containerized
+      type: boolean
+      description: >
+        If the host is a container.
+
+    - name: os.build
+      type: keyword
+      example: "18D109"
+      description: >
+        OS build information.
+
+    - name: os.codename
+      type: keyword
+      example: "stretch"
+      description: >
+        OS codename, if any.
+
+- name: agent
+  title: Agent
+  type: group
+  fields:
+    - name: id
+      type: keyword
+      ignore_above: 1024
+      dimension: true

packages/kafka/data_stream/producer/fields/base-fields.yml

@@ -0,0 +1,20 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.
+- name: event.module
+  type: constant_keyword
+  description: Event module
+  value: kafka
+- name: event.dataset
+  type: constant_keyword
+  description: Event dataset
+  value: kafka.producer

packages/kafka/data_stream/producer/fields/ecs.yml

@@ -0,0 +1,3 @@
+- external: ecs
+  name: service.address
+  dimension: true