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)

Java SDK Usage

The Splunk source has no dedicated builder method; attach it with appendNode("splunksource", config). The credential is registered in the JobContext under a credentialId.

Basic Usage

JobContext jobContext = JobContext.builder()
    .otherProperties(new HashMap<>(Map.of(
        "splunk-cred",
        new UsernamePasswordCredential("splunk-username", "splunk-password"))))
    .build();

ZephFlow flow = ZephFlow.startFlow(jobContext)
    .appendNode("splunksource", SplunkSourceDto.Config.builder()
        .splunkUrl("https://splunk.example.com:8089")
        .searchQuery("search index=main sourcetype=access_log | head 100")
        .validateCertificates(false)
        .credentialId("splunk-cred")
        .build());

Log Ingestion Pipeline

JobContext jobContext = JobContext.builder()
    .otherProperties(new HashMap<>(Map.of(
        "splunk-cred",
        new UsernamePasswordCredential("splunk-user", "splunk-pass"))))
    .build();

ZephFlow flow = ZephFlow.startFlow(jobContext)
    .appendNode("splunksource", SplunkSourceDto.Config.builder()
        .splunkUrl("https://splunk.example.com:8089")
        .searchQuery("search index=main sourcetype=access_log | head 1000")
        .validateCertificates(false)
        .credentialId("splunk-cred")
        .build())
    .eval("dict_merge($, dict(ingested=true))")
    .kafkaSink("kafka:9092", "splunk-records", null, EncodingType.JSON_OBJECT, null);

With SSL Validation

ZephFlow flow = ZephFlow.startFlow(jobContext)
    .appendNode("splunksource", SplunkSourceDto.Config.builder()
        .splunkUrl("https://splunk-prod.internal:8089")
        .searchQuery("search index=security sourcetype=firewall action=blocked")
        .validateCertificates(true)
        .credentialId("splunk-cred")
        .build())
    .filter("$.severity == \"high\"")
    .s3Sink("us-east-1", "security-logs", "blocked-records", EncodingType.JSON_OBJECT_LINE);

Related Nodes

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