Splunk Integration

How to integrate Splunk with Connectware, including setting up an HTTP Event Collector token, streaming shop floor events into a Splunk index, and adding the equipment hierarchy as indexed fields.

This guide describes how to stream shop floor events from Connectware into Splunk through the HTTP Event Collector (HEC). You configure a service commissioning file that subscribes to machine events on the MQTT broker, wraps each event in the HEC JSON envelope, and sends it to a Splunk index. A complete example file is available at the end of this guide.

Objectives

• Setting up an HTTP Event Collector token in Splunk.

• Connecting Connectware to Splunk HEC with token authentication.

• Streaming machine events from an ISA-95-style MQTT topic hierarchy into a Splunk index.

• Adding the equipment hierarchy as indexed fields, so events are searchable by site, area, line, and cell.

Prerequisites

To follow this guide, you will need the following:

• A running instance of Cybus Connectware.

• A running Splunk instance: Splunk Enterprise or Splunk Cloud Platform, with permissions to create an HTTP Event Collector token.

• Access to the Admin UI with sufficient user permissions.

• Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, and endpoints).

Connectware and Splunk Integration

Splunk ingests data over HTTP through the HTTP Event Collector, a token-authenticated receiver for JSON events. Connectware communicates with HEC through the HTTP/REST connector. Every event is sent as a POST request to the /services/collector/event endpoint, wrapped in a JSON envelope that carries the event payload and its metadata, such as the target index, the source type, and the event time. The Authorization: Splunk <token> header authenticates each request.

The HEC address depends on your Splunk deployment:

• Splunk Enterprise: HEC listens on TCP port 8088 of your Splunk server, with TLS enabled by default.

• Splunk Cloud Platform: HEC is reachable on port 443 under a dedicated hostname. On AWS deployments, the pattern is http-inputs-<host>.splunkcloud.com. On Google Cloud, Azure, and AWS GovCloud deployments, the pattern is http-inputs.<host>.splunkcloud.com, where <host> is the name of your Splunk Cloud Platform instance.

The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (<enterprise>/<site>/<area>/<line>/<cell>). The mapping subscribes with wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration. The wildcard values also end up as indexed fields on every event, which keeps the equipment context searchable in Splunk.

Setting Up the HTTP Event Collector

HEC is disabled by default on Splunk Enterprise. To enable it and create a token:

1. In Splunk Web, open Settings > Data Inputs > HTTP Event Collector.

2. Click Global Settings, set All Tokens to Enabled, and click Save. This dialog also shows the HEC port (default: 8088) and whether SSL is enabled.

3. Click New Token and follow the wizard. Give the token a name and select the allowed indexes and the default index.

4. Copy the token value. You need it when you install the service.

On Splunk Cloud Platform, you do not manage global HEC settings. Create the token in the same menu and use the dedicated HEC hostname of your deployment.

Splunk Connection Properties

The connection to Splunk requires the HEC address and the token. We add them as parameters to the service commissioning file, so you can set them when you install the service.

Do not worry about copying the service commissioning file snippets together into one, the complete example file is available at the end of this guide.

• splunkHost: The hostname of your Splunk instance, without the scheme. For Splunk Cloud Platform, use the dedicated HEC hostname, for example http-inputs-example.splunkcloud.com.

• splunkPort: The HEC port. Defaults to 8088 for Splunk Enterprise. Use 443 for Splunk Cloud Platform.

• hecToken: The HEC token that authenticates event submissions.

• splunkIndex: The Splunk index that receives the events. The index must be in the allowed indexes of the token.

• topicRoot: The root of the MQTT topic hierarchy. Defaults to enterprise.

parameters:
  splunkHost:
    description: Hostname of your Splunk instance, without the scheme
    type: string
    default: splunk.example.com

  splunkPort:
    description: HEC port, 8088 for Splunk Enterprise, 443 for Splunk Cloud Platform
    type: number
    default: 8088

  hecToken:
    description: HEC token used to authenticate event submissions
    type: string

  splunkIndex:
    description: Splunk index that receives the events
    type: string
    default: main

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

Splunk Connection

To connect to HEC, we set up a Cybus::Connection resource that uses the HTTP/REST connector. The headers property adds the Authorization: Splunk header to every request, so the endpoints that use this connection do not need any authentication configuration of their own.

The connection probes the /services/collector/health path, which responds without authentication when HEC is enabled and ready to accept events. A connected state therefore confirms the host, the port, and the TLS setup, but not the token.

resources:
  splunkConnection:
    type: Cybus::Connection
    properties:
      protocol: Http
      connection:
        scheme: https
        host: !ref splunkHost
        port: !ref splunkPort
        headers:
          Authorization: !sub 'Splunk ${hecToken}'
        probePath: /services/collector/health
        probeMethod: GET

On Splunk Enterprise, HEC uses a self-signed certificate by default. Connectware cannot validate this certificate, so the connection does not reach the connected state. Either configure HEC with a certificate from your own certificate authority (CA) and add the CA certificate as Base64-encoded PEM content in the caCert property of the connection, or set trustAllCertificates: true to accept any certificate. Use trustAllCertificates only in test environments because it disables server identity verification.

Streaming Machine Events into Splunk

HEC creates one event per JSON envelope that is posted to the /services/collector/event path. We define a write endpoint for this path and a mapping that feeds it from the MQTT topic hierarchy.

The HTTP/REST connector expects the request body in the body property of the message (see Publishing Data to REST Servers). The transform rule builds the HEC envelope inside this body property, so machines can publish their data without knowing about either convention.

eventEndpoint:
  type: Cybus::Endpoint
  properties:
    protocol: Http
    connection: !ref splunkConnection
    write:
      path: /services/collector/event

eventMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: !sub '${topicRoot}/+site/+area/+line/+cell/events'
        publish:
          endpoint: !ref eventEndpoint
        rules:
          - transform:
              expression: !sub |
                {
                  "body": {
                    "event": $,
                    "time": $millis() / 1000,
                    "host": $context.vars.cell,
                    "source": "connectware",
                    "sourcetype": "connectware:event",
                    "index": "${splunkIndex}",
                    "fields": {
                      "site": $context.vars.site,
                      "area": $context.vars.area,
                      "line": $context.vars.line,
                      "cell": $context.vars.cell
                    }
                  }
                }

The envelope properties have the following meaning in Splunk:

• event: The event payload. The transform passes the incoming MQTT message through unchanged.

• time: The event time in epoch seconds. HEC accepts fractional values, so $millis() / 1000 keeps millisecond precision. If your machine payload already carries an event timestamp, use that field instead.

• host, source, and sourcetype: Descriptive metadata for filtering and parsing in Splunk. This example uses the cell level of the topic as the host.

• index: The Splunk index that receives the event. If you omit this property, HEC uses the default index of the token.

• fields: A flat object of custom fields that Splunk indexes directly. The named wildcards of the topic make the equipment hierarchy levels available in $context.vars, so every event is searchable by site, area, line, and cell without any search-time extraction. Indexed fields require the /services/collector/event path, they are not supported for raw data.

Any message published to a matching topic now becomes one Splunk event. For example, a machine stop published to enterprise/hamburg/assembly/line-1/press-01/events:

{
  "eventType": "machine-stop",
  "reason": "Material jam at infeed",
  "operator": "shift-a"
}

This results in an event in the main index with the source type connectware:event, the host press-01, and the indexed fields site=hamburg, area=assembly, line=line-1, and cell=press-01.

Verifying the Integration

1. Install the service and set the parameters with the values of your Splunk instance.

2. Check that the connection is in the Connected state on the service details page in the Admin UI. The probe calls the HEC health endpoint, so a connected state confirms the address and the TLS setup. A wrong token only shows up when the first event is sent.

3. Publish a test message to enterprise/hamburg/assembly/line-1/press-01/events, for example with an MQTT client or the Admin UI.

4. Open the Search & Reporting app in Splunk and run the search index=main sourcetype="connectware:event" over the last 15 minutes. The test message appears as an event, and you can narrow the search with the indexed fields, for example site=hamburg.

5. The result of every HTTP request is published to the /res topic of the endpoint. Use the Data Explorer to inspect it. On success, the result contains the HEC response {"text":"Success","code":0}. If Splunk rejects a request, the message on the /res topic contains an error property with the HTTP status, for example 403 Forbidden for an invalid token.

Service Commissioning File Example

splunk-example.yml

3KB

Open

splunk-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# Splunk Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for streaming shop floor events from
  Connectware to Splunk through the HTTP Event Collector (Example)

metadata:
  name: Splunk Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  splunkHost:
    description: Hostname of your Splunk instance, without the scheme
    type: string
    default: splunk.example.com

  splunkPort:
    description: HEC port, 8088 for Splunk Enterprise, 443 for Splunk Cloud Platform
    type: number
    default: 8088

  hecToken:
    description: HEC token used to authenticate event submissions
    type: string

  splunkIndex:
    description: Splunk index that receives the events
    type: string
    default: main

  topicRoot:
    description: Root of the MQTT topic hierarchy
    type: string
    default: enterprise

resources:
  # Connection to the Splunk HTTP Event Collector
  splunkConnection:
    type: Cybus::Connection
    properties:
      protocol: Http
      connection:
        scheme: https
        host: !ref splunkHost
        port: !ref splunkPort
        headers:
          Authorization: !sub 'Splunk ${hecToken}'
        probePath: /services/collector/health
        probeMethod: GET
        # On Splunk Enterprise, HEC uses a self-signed certificate by default.
        # Provide your CA certificate, or accept any certificate in test
        # environments only:
        # caCert: <Base64-encoded PEM content of your CA certificate>
        # trustAllCertificates: true

  # Sends one HEC event envelope per POST request
  eventEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Http
      connection: !ref splunkConnection
      write:
        path: /services/collector/event

  # Wraps every machine event in the HEC JSON envelope
  eventMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: !sub '${topicRoot}/+site/+area/+line/+cell/events'
          publish:
            endpoint: !ref eventEndpoint
          rules:
            - transform:
                expression: !sub |
                  {
                    "body": {
                      "event": $,
                      "time": $millis() / 1000,
                      "host": $context.vars.cell,
                      "source": "connectware",
                      "sourcetype": "connectware:event",
                      "index": "${splunkIndex}",
                      "fields": {
                        "site": $context.vars.site,
                        "area": $context.vars.area,
                        "line": $context.vars.line,
                        "cell": $context.vars.cell
                      }
                    }
                  }