feat: add opensearch provider

docs/mint.json

@@ -219,6 +219,7 @@
             "providers/documentation/ollama-provider",
             "providers/documentation/openai-provider",
             "providers/documentation/openobserve-provider",
+            "providers/documentation/opensearch-provider",
             "providers/documentation/opensearchserverless-provider",
             "providers/documentation/openshift-provider",
             "providers/documentation/opsgenie-provider",

docs/providers/documentation/opensearch-provider.mdx

@@ -0,0 +1,65 @@
+---
+title: "OpenSearch"
+sidebarTitle: "OpenSearch Provider"
+description: "OpenSearch provider enables Keep to query and write to self-managed OpenSearch clusters."
+---
+
+## Overview
+
+The OpenSearch provider connects Keep to a self-managed OpenSearch cluster. You can run queries, index documents, and surface alerts from OpenSearch monitors into Keep.
+
+### Key Features
+- Query indices with the OpenSearch Query DSL.
+- Index documents (with optional document IDs and refresh policy).
+- TLS and basic authentication support; custom CA bundle (`ca_certs`) for private PKI.
+
+## Prerequisites
+- An accessible OpenSearch endpoint (host and port).
+- If authentication is enabled, a user with query/write permissions to the target indices.
+- For HTTPS with private CAs, a PEM bundle path for certificate verification.
+
+## Authentication Configuration
+- **host** (Required): OpenSearch host (e.g., `opensearch.example.com` or `http://localhost`)
+- **port** (Required): OpenSearch port (default `9200`)
+- **username/password** (Optional): Set when using Basic auth (both must be provided together)
+- **use_ssl** (Optional): Enable HTTPS (default: true)
+- **verify_certs** (Optional): Verify server certificates (default: true)
+- **ca_certs** (Optional): PEM bundle path for custom CAs when using HTTPS
+
+## Querying OpenSearch
+Use the `_query` action to search an index.
+
+- **index**: Target index name.
+- **query**: OpenSearch query DSL object (stringified JSON also supported).
+- **size** (optional): Limit the number of returned documents.
+
+Example:
+```json
+{
+  "index": "keep"
+  "query": { "match_all": {} },
+  "size": 1
+}
+```
+
+## Writing to OpenSearch
+Use the `_notify` action to index documents.
+
+- **index**: Target index name.
+- **document**: JSON object (or stringified JSON) to index.
+- **doc_id** (optional): Explicit document ID.
+- **refresh** (optional): Refresh policy (`true`, `false`, or `"wait_for"`).
+
+Example:
+```json
+{
+  "index": "keep",
+  "document": { "message": "Keep test doc" },
+  "doc_id": "doc_1",
+  "refresh": true
+}
+```
+
+## Useful Links
+- [OpenSearch Query DSL](https://opensearch.org/docs/latest/query-dsl/index/)
+- [Index API](https://opensearch.org/docs/latest/api-reference/document-apis/index-document/)

docs/providers/overview.mdx

@@ -596,6 +596,14 @@ By leveraging Keep Providers, users are able to deeply integrate Keep with the t
   }
 ></Card>
 
+<Card
+  title="OpenSearch"
+  href="/providers/documentation/opensearch-provider"
+  icon={
+    <img src="https://img.logo.dev/opensearch.com?token=pk_dfXfZBoKQMGDTIgqu7LvYg" >
+  }
+></Card>
+
 <Card
   title="OpenSearch Serverless"
   href="/providers/documentation/opensearchserverless-provider"

docs/snippets/providers/opensearch-snippet-autogenerated.mdx

@@ -0,0 +1,55 @@
+{/* This snippet is automatically generated using scripts/docs_render_provider_snippets.py
+Do not edit it manually, as it will be overwritten */}
+
+## Authentication
+This provider requires authentication.
+- **host**: OpenSearch host (required: True, sensitive: False)
+- **port**: OpenSearch port (required: True, sensitive: False)
+- **username**: OpenSearch username (required: False, sensitive: False)
+- **password**: OpenSearch password (required: False, sensitive: True)
+- **use_ssl**: Enable SSL (required: False, sensitive: False)
+- **verify_certs**: Enable SSL certificate verification (required: False, sensitive: False)
+- **ca_certs**: CA bundle path (PEM) for HTTPS connections (required: False, sensitive: False)
+
+Certain scopes may be required to perform specific actions or queries via the provider. Below is a summary of relevant scopes and their use cases:
+- **connect_to_server**: The user can connect to the server (mandatory) 
+
+
+
+## In workflows
+
+This provider can be used in workflows.
+
+
+As "step" to query data, example:
+```yaml
+steps:
+    - name: Query opensearch
+      provider: opensearch
+      config: "{{ provider.my_provider_name }}"
+      with:
+        query: {value}  
+        index: {value}  
+        size: {value}  
+```
+
+
+As "action" to make changes or update data, example:
+```yaml
+actions:
+    - name: Query opensearch
+      provider: opensearch
+      config: "{{ provider.my_provider_name }}"
+      with:
+        index: {value}  
+        document: {value}  
+        doc_id: {value}  
+        refresh: {value}  
+```
+
+
+
+
+Check the following workflow examples:
+- [opensearch_basic.yml](https://github.com/keephq/keep/blob/main/examples/workflows/opensearch_basic.yml)
+- [opensearchserverless_basic.yml](https://github.com/keephq/keep/blob/main/examples/workflows/opensearchserverless_basic.yml)

examples/workflows/opensearch_basic.yml

@@ -0,0 +1,33 @@
+workflow:
+  id: opensearch-create-query
+  name: OSS OpenSearch Create & Query Docs
+  description: >
+    Retrieves all the documents from index "keep",
+    and uploads a document to OSS OpenSearch in index "keep".
+  disabled: false
+
+  triggers:
+    - type: manual
+
+  steps:
+    - name: query-index
+      provider:
+        type: opensearch
+        config: "{{ providers.opensearch }}"
+        with:
+          index: keep
+          query:
+            match_all: {}
+          size: 1
+
+  actions:
+    - name: create-doc
+      provider:
+        type: opensearch
+        config: "{{ providers.opensearch }}"
+        with:
+          index: keep
+          document:
+            message: Keep test doc
+          doc_id: doc_1
+          refresh: true

keep/providers/opensearch_provider/opensearch_provider.py

@@ -0,0 +1,198 @@
+import dataclasses
+import json
+import typing
+
+import pydantic
+from opensearchpy import OpenSearch
+
+from keep.contextmanager.contextmanager import ContextManager
+from keep.exceptions.provider_connection_failed import ProviderConnectionFailed
+from keep.exceptions.provider_exception import ProviderException
+from keep.providers.base.base_provider import BaseProvider
+from keep.providers.models.provider_config import ProviderConfig, ProviderScope
+
+
+@pydantic.dataclasses.dataclass
+class OpensearchProviderAuthConfig:
+    host: str = dataclasses.field(
+        default=None,
+        metadata={
+            "required": True,
+            "description": "OpenSearch host",
+            "sensitive": False,
+        },
+    )
+    port: str = dataclasses.field(
+        default=9200,
+        metadata={
+            "required": True,
+            "description": "OpenSearch port",
+            "sensitive": False,
+        },
+    )
+    username: typing.Optional[str] = dataclasses.field(
+        default=None,
+        metadata={
+            "description": "OpenSearch username",
+            "config_sub_group": "username_password",
+            "config_main_group": "authentication",
+            "sensitive": False,
+        },
+    )
+    password: typing.Optional[str] = dataclasses.field(
+        default=None,
+        metadata={
+            "description": "OpenSearch password",
+            "config_sub_group": "username_password",
+            "config_main_group": "authentication",
+            "sensitive": True,
+        },
+    )
+    use_ssl: bool = dataclasses.field(
+        default=True,
+        metadata={
+            "description": "Enable SSL",
+            "type": "switch",
+            "config_main_group": "authentication",
+        },
+    )
+    verify_certs: bool = dataclasses.field(
+        default=True,
+        metadata={
+            "description": "Enable SSL certificate verification",
+            "type": "switch",
+            "config_main_group": "authentication",
+        },
+    )
+    ca_certs: typing.Optional[str] = dataclasses.field(
+        default=None,
+        metadata={
+            "description": "CA bundle path (PEM) for HTTPS connections",
+            "config_main_group": "authentication",
+            "config_sub_group": "ssl",
+            "placeholder": "/path/to/ca.pem",
+        },
+    )
+
+
+class OpensearchProvider(BaseProvider):
+    PROVIDER_DISPLAY_NAME = "OpenSearch"
+    PROVIDER_CATEGORY = ["Database", "Observability"]
+
+    PROVIDER_SCOPES = [
+        ProviderScope(
+            name="connect_to_server",
+            description="The user can connect to the server",
+            mandatory=True,
+            alias="Connect to the server",
+        )
+    ]
+
+    def __init__(
+        self, context_manager: ContextManager, provider_id: str, config: ProviderConfig
+    ):
+        super().__init__(context_manager, provider_id, config)
+        self._client: OpenSearch | None = None
+
+    def dispose(self):
+        pass
+
+    def validate_config(self):
+        self.authentication_config = OpensearchProviderAuthConfig(
+            **self.config.authentication
+        )
+
+    def validate_scopes(self):
+        try:
+            self.client.ping()
+            scopes = {
+                "connect_to_server": True,
+            }
+        except Exception as e:
+            self.logger.exception("Error validating scopes")
+            scopes = {
+                "connect_to_server": str(e),
+            }
+        return scopes
+
+    @property
+    def client(self) -> OpenSearch:
+        if not self._client:
+            self._client = self.__initialize_client()
+        return self._client
+
+    def __initialize_client(self) -> OpenSearch:
+        scheme = "https" if self.authentication_config.use_ssl else "http"
+        auth = None
+        if self.authentication_config.username:
+            auth = (
+                self.authentication_config.username,
+                self.authentication_config.password,
+            )
+        client = OpenSearch(
+            hosts=[
+                {
+                    "host": self.authentication_config.host,
+                    "port": self.authentication_config.port,
+                    "scheme": scheme,
+                }
+            ],
+            http_auth=auth,
+            use_ssl=self.authentication_config.use_ssl,
+            verify_certs=self.authentication_config.verify_certs,
+            ca_certs=self.authentication_config.ca_certs,
+        )
+        try:
+            client.info()
+        except Exception as e:
+            raise ProviderConnectionFailed(f"Failed to connect to OpenSearch: {str(e)}")
+        return client
+
+    def _query(self, query: dict | str, index: str, size: int | None = None, **kwargs):
+        if not index:
+            raise ProviderException("Missing index for OpenSearch query")
+        body = query
+        if isinstance(body, str):
+            body = json.loads(body)
+        if size is not None:
+            if not isinstance(body, dict):
+                raise ProviderException("Query must be an object when specifying size")
+            body = dict(body)
+            body["size"] = size
+        response = self.client.search(index=index, body=body)
+        hits = response.get("hits", {}).get("hits")
+        return hits or []
+
+    def _notify(
+        self,
+        index: str,
+        document: dict | str,
+        doc_id: str | None = None,
+        refresh: bool | str | None = None,
+        **kwargs,
+    ):
+        if not index:
+            raise ProviderException("Missing index for OpenSearch document")
+        if document is None:
+            raise ProviderException("Missing document for OpenSearch")
+        body = document
+        if isinstance(body, str):
+            body = json.loads(body)
+        params: dict[str, typing.Any] = {
+            "index": index,
+            "body": body,
+        }
+        if doc_id:
+            params["id"] = doc_id
+        if refresh is not None:
+            params["refresh"] = refresh
+        return self.client.index(**params)
+
+    def get_provider_metadata(self) -> dict:
+        try:
+            info = self.client.info()
+            version = info.get("version", {}).get("number")
+            return {"version": version} if version else {"version": "unknown"}
+        except Exception:
+            self.logger.exception("Failed to get OpenSearch metadata")
+            return {}