Adding Splunk as a New Ingestion Source

[drona-gyawali]

Hi @Lorygold, @ManofWax,

While reviewing ingestion.json, I noticed a reference to Splunk. Given its widespread use, I believe adding it as a new ingestion source in BuffaLogs would be valuable. This would allow users to pull logs directly from Splunk and integrate them into BuffaLogs for further analysis.

Approach: Using Splunk's REST API (/services/search/jobs/export)

Instead of relying on ad-hoc queries, I propose implementing an ingestion mechanism using Splunk's REST API—specifically the /services/search/jobs/export endpoint. This method allows us to fetch logs efficiently without managing long-running search jobs.

Steps to Implement:

1. Follow the BaseIngestionConnector Structure – Since BuffaLogs has a standardized ingestion framework, the Splunk connector will extend BaseIngestionConnector, ensuring consistency.

2. API Integration – Implement a REST client to:

       - Authenticate with Splunk (API Key / Basic Auth).
   
       - Execute log retrieval queries using /services/search/jobs/export.
   
       - Parse and transform results into BuffaLogs’ expected format.
   

3. Write Unit Tests – Cover API interactions and data transformation logic.

4. Develop Documentation – Provide configuration instructions for users.

This approach keeps the ingestion process in line with BuffaLogs' existing architecture. If you have any suggestions or considerations that I should keep in mind, I’d love to hear your thoughts!

[ManofWax]

Yes, developing a Splunk Ingestion could be very useful.

I'm not very familiar with Splunk APIs: your intent is using /services/search/jobs/export to query for every login found in a timerange and process them one by one?

In our Elasticsearch implementation we use two functions:

process_users(): query elasticsearch and aggregates the results for username (basically it returns a list of unique usernames seen). In this way the query is very fast even if we need to process a lot of logins.

process_user_logins(): will process usernames one-by-one and query every login seen for each of them.

If it is possible a similar approach should be used also for splunk

[drona-gyawali]

Yes, the approach mirrors Elasticsearch's two-step process:

1. process_users:

   • Uses a Splunk query with dedup to fetch unique usernames with successful logins in the timerange.
   • Example query:

     search index=* ... | dedup user_name | table user_name
     

   • This avoids processing all events upfront, just like Elasticsearch's aggregation.

2. process_user_logins:

   • For each username, queries all their login events (with geo/IP details) in the timerange.
   • Example query:

     search index=* user_name="alice" ... | table _time source_ip ...
     

   • Results are normalized into BuffaLogs' format using normalize_fields.

Example

def process_users(self, start_date, end_date):
        """Get unique users with successful logins in time range"""
        start_epoch = int(start_date.timestamp())
        end_epoch = int(end_date.timestamp())

        query = (
            f"search index=* earliest={start_epoch} latest={end_epoch} "
            'event_category="authentication" event_outcome="success" '
            'event_type="start" user_name=* '
            f"| dedup user_name | table user_name"
        )

        self.logger.info(f"Executing user query: {query}")
        results = self._execute_splunk_query(query)
        return [user.get("user_name") for user in results if user.get("user_name")]

def process_user_logins(self, start_date, end_date, username):
        """Get all login events for a specific user"""
        start_epoch = int(start_date.timestamp())
        end_epoch = int(end_date.timestamp())

        query = (
            f"search index=* earliest={start_epoch} latest={end_epoch} "
            f'user_name="{username}" event_category="authentication" '
            'event_outcome="success" event_type="start" source_ip=* '
            "| table _time source_ip user_agent source_geo_location_lat "
            "source_geo_location_lon source_geo_country_name "
            "source_as_organization_name _index _id"
        )

        self.logger.info(f"Executing login query for {username}: {query}")
        return self._execute_splunk_query(query)

Additionally, we have an option of using the Splunk SDK as well, which can be super fast and easy to configure. Please let me know what you think. And, I have some doubts regarding the schema and normalization. Would love to hear your thoughts on this.

Thank you!

[drona-gyawali]

Hey @ManofWax, @Lorygold

I was doing some research regarding Splunk ingestion and worked on this as part of my learning process so this isn’t PR-ready code yet. I’ve implemented the test case and Splunk ingestion logic here: commit link

I’m a bit confused do we need to use ECS for Splunk ingestion, or is there another approach? If ECS is required, then I tried to align Splunk's flat key-value schema with the dot-nested structure used in Elasticsearch, but I came up with a workaround for it in above commit

Could you review it and let me know if this approach makes sense or if there's a better way to handle the normalization?

Looking forward to your feedback!

Thanks!

[ManofWax]

I think your code looks good. In Buffalogs we choose fields names resembling ECS but in future we could diverge from ECS schema, that's why we implemented a normalization function on every ingestion sources. Your implementation looks correct. As Lorena said you can go on and make a PR as soon as your code is ready.

[Lorygold]

Hi @drona-gyawali, you can open an issue using the [FEATURE]: Integrating <source_name> ingestion source template, so we can assign you to this feature development.
Complete the sections required there and explain your doubts in the in the relevant division

[Lorygold]

Discussion closed, because the related issue has been opened: #220