Onboarding OpenSearch as a New Ingestion Source for BuffaLogs

[rskbansal]

BuffaLogs currently supports Elasticsearch as its sole data ingestion source. As part of my GSoC project, I am expanding its capabilities to support additional data sources, with OpenSearch being one of the first major additions.

OpenSearch, an open-source, community-driven fork of Elasticsearch, is a really similar search and analytics engine. It is widely used by organizations worldwide as a scalable and flexible alternative to Elasticsearch, particularly after Elastic changed its licensing model. So it makes sense to onboard it in parallel to Elasticsearch.

I already made a docker-compose.opensearch.yaml file to create a local instance of OpenSearch to simplify testing, which looks something of this sort:

services:
    opensearch:
        container_name: buffalogs_opensearch
        image: opensearchproject/opensearch:latest
        hostname: opensearch
        ports:
            - "9200:9200"
            - "9600:9600"

    opensearch-dashboards:
        container_name: buffalogs_opensearch-dashboards
        image: opensearchproject/opensearch-dashboards:latest
        hostname: opensearch-dashboards
        ports:
            - "127.0.0.1:5601:5601"

I have also created a new OpenSearchIngestion class that is similar to how Elasticsearch is implemented. Here's a code snippet for the same with an example of the process_users() function, which can be extended to other functions:

from opensearchpy import OpenSearch
from impossible_travel.ingestion.base_ingestion import BaseIngestion

class OpenSearchIngestion(BaseIngestion):
    def __init__(self, ingestion_config: dict):
        super().__init__()
        self.opensearch_config = ingestion_config
        self.client = OpenSearch(
            hosts=[self.opensearch_config["url"]],
            timeout=self.opensearch_config["timeout"],
            verify_certs=False,
        )

    def process_users(self, start_date, end_date) -> list:
        users_list = []
        query = {
            "query": {
                "bool": {
                    "must": [
                        {"range": {"@timestamp": {"gte": start_date, "lt": end_date}}},
                        {"match": {"event.category": "authentication"}},
                        {"match": {"event.outcome": "success"}},
                        {"match": {"event.type": "start"}},
                        {"exists": {"field": "user.name"}},
                    ]
                }
            },
            "aggs": {
                "login_user": {
                    "terms": {"field": "user.name", "size": self.opensearch_config["bucket_size"]}
                }
            }
        }
        response = self.client.search(index=self.opensearch_config["indexes"], body=query)
        for user in response['aggregations']['login_user']['buckets']:
            users_list.append(user['key'])
        return users_list

Before opening a Pull Request, I’d love to hear your feedback on the implementation—any suggestions, improvements, or potential issues you foresee.

[rskbansal]

@Lorygold may I have your views on this.

[Lorygold]

Discussion closed because the task has been accepted → related issue: #202