Splunk Source Node

The splunkSource node in ZephFlow enables you to ingest data from Splunk Enterprise or Splunk Cloud Platform by executing SPL search queries against the Splunk REST API.

Overview

The Splunk source node connects to a Splunk instance via its management port, submits a search job using SPL (Search Processing Language), waits for the job to complete, and emits the resulting records into your ZephFlow pipeline. This is a batch source — it runs the query once per pipeline execution.

Key Features

• SPL Query Support: Execute any valid SPL search query against your Splunk instance
• Credential Management: Authenticate with username/password credentials
• SSL Configuration: Optional SSL certificate validation for HTTPS connections

Parameters

Parameter	Type	Description	Required
splunkUrl	String	Full URL of the Splunk management endpoint (e.g., https://host:8089)	Yes
searchQuery	String	SPL query to execute (e.g., search index=main sourcetype=syslog)	Yes
validateCertificates	boolean	Whether to validate SSL certificates for HTTPS connections	No
credentialId	String	ID of a UsernamePasswordCredential in jobContext.otherProperties	Yes

Config Object

The full config object for the Splunk source node (SplunkSourceDto.Config):

Field	Type	Required	Default	Description
splunkUrl	String	Yes	—	Full URL of the Splunk management endpoint (e.g., https://host:8089)
searchQuery	String	Yes	—	SPL query to execute
credentialId	String	Yes	—	Credential ID for Splunk authentication
validateCertificates	boolean	No	false	Whether to validate SSL certificates for HTTPS connections
batchSize	int	No	10000	Number of results to fetch per batch
jobInitTimeoutMs	long	No	300000	Timeout in milliseconds for search job initialization
earliestTime	String	No	null	Earliest time bound for the search (Splunk time format)
latestTime	String	No	null	Latest time bound for the search (Splunk time format)

DAG Examples

The credential is registered in jobContext.otherProperties under a credentialId.

Basic Usage

jobContext:
  otherProperties:
    splunk-cred:
      username: "splunk-username"
      password: "splunk-password"

dag:
  - id: "splunk_source"
    commandName: "splunksource"
    config:
      splunkUrl: "https://splunk.example.com:8089"
      searchQuery: "search index=main sourcetype=access_log | head 100"
      validateCertificates: false
      credentialId: "splunk-cred"
    outputs: []

Log Ingestion Pipeline

jobContext:
  otherProperties:
    splunk-cred:
      username: "splunk-user"
      password: "splunk-pass"

dag:
  - id: "splunk_source"
    commandName: "splunksource"
    config:
      splunkUrl: "https://splunk.example.com:8089"
      searchQuery: "search index=main sourcetype=access_log | head 1000"
      validateCertificates: false
      credentialId: "splunk-cred"
    outputs:
      - "mark_ingested"

  - id: "mark_ingested"
    commandName: "eval"
    config:
      expression: "dict_merge($, dict(ingested=true))"
    outputs:
      - "kafka_sink"

  - id: "kafka_sink"
    commandName: "kafkasink"
    config:
      broker: "kafka:9092"
      topic: "splunk-records"
      encodingType: "JSON_OBJECT"
    outputs: []

With SSL Validation

jobContext:
  otherProperties:
    splunk-cred:
      username: "splunk-user"
      password: "splunk-pass"

dag:
  - id: "splunk_source"
    commandName: "splunksource"
    config:
      splunkUrl: "https://splunk-prod.internal:8089"
      searchQuery: "search index=security sourcetype=firewall action=blocked"
      validateCertificates: true
      credentialId: "splunk-cred"
    outputs:
      - "high_severity"

  - id: "high_severity"
    commandName: "filter"
    config:
      expression: '$.severity == "high"'
    outputs:
      - "s3_sink"

  - id: "s3_sink"
    commandName: "s3sink"
    config:
      regionStr: "us-east-1"
      bucketName: "security-logs"
      keyName: "blocked-records"
      encodingType: "JSON_OBJECT_LINE"
    outputs: []

Related Nodes

• kafkaSource: Consume streaming data from Kafka
• s3Sink: Write data to S3