feat: Add Elasticsearch connector integration

- Updated ElasticsearchIndexer to use connector_config for initialization.
- Created a new page for adding Elasticsearch connectors with a comprehensive form.
- Enhanced the dashboard to include the Elasticsearch connector in the connectors list.
- Updated breadcrumb component to display Elasticsearch connector label.
- Added Elasticsearch icon to the connector icons enum.
- Installed @radix-ui/react-radio-group for radio button functionality in the form.

surfsense_backend/app/connectors/elasticsearch_connector.py

@@ -31,35 +31,47 @@ class ElasticsearchConnector:
     async def connect(self) -> bool:
         """Establish connection to Elasticsearch"""
         try:
+            # Build the URL with proper scheme
+            scheme = "https" if self.config.get("ssl_enabled", True) else "http"
+            port = self.config.get("port", 443 if scheme == "https" else 9200)
+            hostname = self.config["hostname"]
+
+            # Construct the full URL
+            url = f"{scheme}://{hostname}:{port}"
+
             connection_params = {
-                "hosts": [f"{self.config.hostname}:{self.config.port}"],
-                "verify_certs": self.config.ssl_enabled
-                if hasattr(self.config, "ssl_enabled")
-                else True,
+                "hosts": [url],
+                "verify_certs": self.config.get("ssl_enabled", True),
                 "request_timeout": 30,
             }
 
-            # Add authentication if provided
+            # Handle different authentication methods
+            auth_method = self.config.get("auth_method", "none")
+
             if (
-                hasattr(self.config, "username")
-                and hasattr(self.config, "password")
-                and self.config.username
-                and self.config.password
+                auth_method == "basic"
+                and self.config.get("username")
+                and self.config.get("password")
             ):
                 connection_params["basic_auth"] = (
-                    self.config.username,
-                    self.config.password,
+                    self.config["username"],
+                    self.config["password"],
                 )
+            elif auth_method == "api_key" and self.config.get("api_key"):
+                api_key = self.config["api_key"]
+                if api_key.startswith("ApiKey "):
+                    connection_params["headers"] = {"Authorization": api_key}
+                else:
+                    connection_params["api_key"] = api_key
 
-            # Add API key authentication if provided
-            if hasattr(self.config, "api_key") and self.config.api_key:
-                connection_params["api_key"] = self.config.api_key
-
+            logger.info(f"Connecting to Elasticsearch at {url}")
             self.client = AsyncElasticsearch(**connection_params)
 
             # Test connection
-            await self.client.info()
-            logger.info("Successfully connected to Elasticsearch")
+            info = await self.client.info()
+            logger.info(
+                f"Successfully connected to Elasticsearch cluster: {info.get('cluster_name', 'Unknown')}"
+            )
             return True
 
         except (ConnectionError, AuthenticationException) as e:
@@ -92,6 +104,9 @@ class ElasticsearchConnector:
                 "cluster_name": info.get("cluster_name"),
                 "version": info.get("version", {}).get("number"),
                 "indices_count": len(indices) if indices else 0,
+                "indices": [
+                    idx["index"] for idx in indices if not idx["index"].startswith(".")
+                ],
             }
         except Exception as e:
             return {"success": False, "error": str(e)}

surfsense_backend/app/routes/__init__.py

@@ -5,7 +5,6 @@ from .airtable_add_connector_route import (
 )
 from .chats_routes import router as chats_router
 from .documents_routes import router as documents_router
-from .elasticsearch_add_connector_route import router as elasticsearch_router
 from .google_calendar_add_connector_route import (
     router as google_calendar_add_connector_router,
 )
@@ -25,7 +24,6 @@ router.include_router(search_spaces_router)
 router.include_router(documents_router)
 router.include_router(podcasts_router)
 router.include_router(chats_router)
-router.include_router(elasticsearch_router)
 router.include_router(search_source_connectors_router)
 router.include_router(google_calendar_add_connector_router)
 router.include_router(google_gmail_add_connector_router)

surfsense_backend/app/routes/elasticsearch_add_connector_route.py

@@ -1,392 +0,0 @@
-import contextlib
-import logging
-from typing import Any
-
-from fastapi import APIRouter, Depends, HTTPException, status
-from pydantic import BaseModel
-from sqlalchemy import select
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.ext.asyncio import AsyncSession
-
-from app.connectors.elasticsearch_connector import ElasticsearchConnector
-from app.db import (
-    SearchSourceConnector,
-    SearchSourceConnectorType,
-    User,
-    get_async_session,
-)
-from app.users import current_active_user
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter(prefix="/elasticsearch", tags=["elasticsearch"])
-
-
-class ElasticsearchConnectorConfig(BaseModel):
-    hostname: str
-    port: int = 9200
-    username: str | None = None
-    password: str | None = None
-    api_key: str | None = None
-    ssl_enabled: bool = True
-    indices: list[str] | None = None
-    query: str = "*"
-    search_fields: list[str] | None = None
-    max_documents: int = 1000
-
-
-@router.post("/add-connector")
-async def add_elasticsearch_connector(
-    connector_data: ElasticsearchConnectorConfig,
-    current_user: User = Depends(current_active_user),
-    session: AsyncSession = Depends(get_async_session),
-) -> dict[str, Any]:
-    """Add a new Elasticsearch connector for the current user"""
-
-    try:
-        elasticsearch_connector = ElasticsearchConnector(connector_data)
-        connection_test = await elasticsearch_connector.test_connection()
-        if not connection_test.get("success"):
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"Failed to connect to Elasticsearch: {connection_test.get('error')}",
-            )
-    except Exception as e:
-        logger.error(f"Error initializing or testing Elasticsearch connector: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Failed to initialize or test Elasticsearch connector: {e!s}",
-        ) from e
-    finally:
-        with contextlib.suppress(Exception):
-            await elasticsearch_connector.disconnect()
-
-    try:
-        # Check if connector already exists for this user
-        existing_connector_result = await session.execute(
-            select(SearchSourceConnector).filter(
-                SearchSourceConnector.user_id == current_user.id,
-                SearchSourceConnector.connector_type
-                == SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
-            )
-        )
-        existing_connector = existing_connector_result.scalars().first()
-
-        if existing_connector:
-            # Update existing connector
-            existing_connector.config = connector_data.model_dump()
-            existing_connector.name = (
-                f"Elasticsearch - {connector_data.hostname}:{connector_data.port}"
-            )
-            existing_connector.is_indexable = True
-            logger.info(
-                f"Updated existing Elasticsearch connector for user {current_user.id}"
-            )
-        else:
-            # Create new connector
-            new_connector = SearchSourceConnector(
-                name=f"Elasticsearch - {connector_data.hostname}:{connector_data.port}",
-                connector_type=SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
-                is_indexable=True,
-                config=connector_data.model_dump(),
-                user_id=current_user.id,
-            )
-            session.add(new_connector)
-            logger.info(
-                f"Created new Elasticsearch connector for user {current_user.id}"
-            )
-
-        await session.commit()
-
-        # Get the connector (either existing or new) to return
-        if existing_connector:
-            connector = existing_connector
-        else:
-            await session.refresh(new_connector)
-            connector = new_connector
-
-        logger.info(
-            f"Successfully saved Elasticsearch connector for user {current_user.id}"
-        )
-
-        return {
-            "success": True,
-            "message": "Elasticsearch connector added successfully",
-            "connector": {
-                "id": connector.id,
-                "name": connector.name,
-                "connector_type": connector.connector_type.value,
-                "is_indexable": connector.is_indexable,
-                "created_at": connector.created_at.isoformat()
-                if connector.created_at
-                else None,
-                "last_indexed_at": connector.last_indexed_at.isoformat()
-                if connector.last_indexed_at
-                else None,
-            },
-        }
-
-    except IntegrityError as e:
-        await session.rollback()
-        raise HTTPException(
-            status_code=status.HTTP_409_CONFLICT,
-            detail=f"Integrity error: A connector with this configuration already exists. {e!s}",
-        ) from e
-    except Exception as e:
-        logger.error(f"Failed to create Elasticsearch connector: {e!s}")
-        await session.rollback()
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to create Elasticsearch connector: {e!s}",
-        ) from e
-
-
-@router.get("/connectors")
-async def get_elasticsearch_connectors(
-    current_user: User = Depends(current_active_user),
-    session: AsyncSession = Depends(get_async_session),
-) -> dict[str, Any]:
-    """Get all Elasticsearch connectors for the current user"""
-
-    try:
-        result = await session.execute(
-            select(SearchSourceConnector).filter(
-                SearchSourceConnector.user_id == current_user.id,
-                SearchSourceConnector.connector_type
-                == SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
-            )
-        )
-        connectors = result.scalars().all()
-
-        connector_list = []
-        for connector in connectors:
-            connector_list.append(
-                {
-                    "id": connector.id,
-                    "name": connector.name,
-                    "connector_type": connector.connector_type.value,
-                    "is_indexable": connector.is_indexable,
-                    "created_at": connector.created_at.isoformat()
-                    if connector.created_at
-                    else None,
-                    "last_indexed_at": connector.last_indexed_at.isoformat()
-                    if connector.last_indexed_at
-                    else None,
-                    "config": {
-                        "hostname": connector.config.get("hostname"),
-                        "port": connector.config.get("port"),
-                        "indices": connector.config.get("indices"),
-                        # Don't expose sensitive credentials
-                    },
-                }
-            )
-
-        return {"success": True, "connectors": connector_list}
-
-    except Exception as e:
-        logger.error(f"Error fetching Elasticsearch connectors: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to fetch Elasticsearch connectors",
-        ) from e
-
-
-@router.put("/connectors/{connector_id}")
-async def update_elasticsearch_connector(
-    connector_id: int,
-    connector_data: ElasticsearchConnectorConfig,
-    current_user: User = Depends(current_active_user),
-    session: AsyncSession = Depends(get_async_session),
-) -> dict[str, Any]:
-    """Update an existing Elasticsearch connector"""
-
-    try:
-        # Get the connector
-        result = await session.execute(
-            select(SearchSourceConnector).filter(
-                SearchSourceConnector.id == connector_id,
-                SearchSourceConnector.user_id == current_user.id,
-                SearchSourceConnector.connector_type
-                == SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
-            )
-        )
-        connector = result.scalar_one_or_none()
-
-        if not connector:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Elasticsearch connector not found",
-            )
-
-        # Test the new connection with robust error handling
-        try:
-            elasticsearch_connector = ElasticsearchConnector(connector_data)
-            connection_test = await elasticsearch_connector.test_connection()
-            if not connection_test.get("success"):
-                raise HTTPException(
-                    status_code=status.HTTP_400_BAD_REQUEST,
-                    detail=f"Failed to connect to Elasticsearch: {connection_test.get('error')}",
-                )
-        except Exception as e:
-            logger.error(f"Error initializing or testing Elasticsearch connector: {e}")
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail=f"Failed to initialize or test Elasticsearch connector: {e!s}",
-            ) from e
-        finally:
-            with contextlib.suppress(Exception):
-                await elasticsearch_connector.disconnect()
-
-        # Update the connector
-        connector.config = connector_data.model_dump()
-        connector.name = (
-            f"Elasticsearch - {connector_data.hostname}:{connector_data.port}"
-        )
-
-        await session.commit()
-        await session.refresh(connector)
-
-        return {
-            "success": True,
-            "message": "Elasticsearch connector updated successfully",
-            "connector": {
-                "id": connector.id,
-                "name": connector.name,
-                "connector_type": connector.connector_type.value,
-                "is_indexable": connector.is_indexable,
-                "created_at": connector.created_at.isoformat()
-                if connector.created_at
-                else None,
-                "last_indexed_at": connector.last_indexed_at.isoformat()
-                if connector.last_indexed_at
-                else None,
-            },
-        }
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        logger.error(f"Error updating Elasticsearch connector: {e}")
-        await session.rollback()
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to update Elasticsearch connector",
-        ) from e
-
-
-@router.delete("/connectors/{connector_id}")
-async def delete_elasticsearch_connector(
-    connector_id: int,
-    current_user: User = Depends(current_active_user),
-    session: AsyncSession = Depends(get_async_session),
-) -> dict[str, Any]:
-    """Delete an Elasticsearch connector"""
-
-    try:
-        # Get the connector
-        result = await session.execute(
-            select(SearchSourceConnector).filter(
-                SearchSourceConnector.id == connector_id,
-                SearchSourceConnector.user_id == current_user.id,
-                SearchSourceConnector.connector_type
-                == SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
-            )
-        )
-        connector = result.scalar_one_or_none()
-
-        if not connector:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail="Elasticsearch connector not found",
-            )
-
-        # Delete the connector
-        await session.delete(connector)
-        await session.commit()
-
-        return {
-            "success": True,
-            "message": "Elasticsearch connector deleted successfully",
-        }
-
-    except HTTPException:
-        raise
-    except Exception as e:
-        logger.error(f"Error deleting Elasticsearch connector: {e}")
-        await session.rollback()
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to delete Elasticsearch connector",
-        ) from e
-
-
-@router.get("/test-connection")
-async def test_elasticsearch_connection(
-    hostname: str,
-    port: int = 9200,
-    username: str | None = None,
-    password: str | None = None,
-    api_key: str | None = None,
-    ssl_enabled: bool = True,
-    current_user: User = Depends(current_active_user),
-) -> dict[str, Any]:
-    """Test Elasticsearch connection with provided credentials"""
-
-    config_data = ElasticsearchConnectorConfig(
-        hostname=hostname,
-        port=port,
-        username=username,
-        password=password,
-        api_key=api_key,
-        ssl_enabled=ssl_enabled,
-    )
-
-    elasticsearch_connector = ElasticsearchConnector(config_data)
-
-    try:
-        result = await elasticsearch_connector.test_connection()
-        return result
-    except Exception as e:
-        logger.error(f"Error testing Elasticsearch connection: {e}")
-        return {"success": False, "error": str(e)}
-    finally:
-        await elasticsearch_connector.disconnect()
-
-
-@router.get("/indices")
-async def get_elasticsearch_indices(
-    hostname: str,
-    port: int = 9200,
-    username: str | None = None,
-    password: str | None = None,
-    api_key: str | None = None,
-    ssl_enabled: bool = True,
-    current_user: User = Depends(current_active_user),
-) -> dict[str, Any]:
-    """Get list of available Elasticsearch indices"""
-
-    config_data = ElasticsearchConnectorConfig(
-        hostname=hostname,
-        port=port,
-        username=username,
-        password=password,
-        api_key=api_key,
-        ssl_enabled=ssl_enabled,
-    )
-
-    elasticsearch_connector = ElasticsearchConnector(config_data)
-
-    try:
-        if not await elasticsearch_connector.connect():
-            return {"success": False, "error": "Failed to connect to Elasticsearch"}
-
-        indices = await elasticsearch_connector.get_indices()
-        return {"success": True, "indices": indices}
-    except Exception as e:
-        logger.error(f"Error fetching Elasticsearch indices: {e}")
-        return {"success": False, "error": str(e)}
-    finally:
-        await elasticsearch_connector.disconnect()
-
-
-# Note: Removed the index_elasticsearch_connector endpoint - now handled by universal system

surfsense_backend/app/schemas/search_source_connector.py

@@ -216,6 +216,7 @@ class SearchSourceConnectorBase(BaseModel):
                 "username",
                 "password",
                 "api_key",
+                "auth_method",
                 "ssl_enabled",
                 "indices",
                 "query",
@@ -227,6 +228,18 @@ class SearchSourceConnectorBase(BaseModel):
             if not config.get("hostname"):
                 raise ValueError("Elasticsearch connector must have a hostname")
 
+            # Validate authentication method
+            auth_method = config.get("auth_method", "none")
+            if auth_method not in ["none", "basic", "api_key"]:
+                raise ValueError("auth_method must be one of: none, basic, api_key")
+
+            # Validate auth credentials based on method
+            if auth_method == "basic":
+                if not config.get("username") or not config.get("password"):
+                    raise ValueError("Username and password required for basic auth")
+            elif auth_method == "api_key" and not config.get("api_key"):
+                raise ValueError("API key required for api_key auth method")
+
             # Validate that all config keys are allowed
             for key in config:
                 if key not in allowed_keys:

surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py

@@ -32,7 +32,7 @@ class ElasticsearchIndexer:
 
     def __init__(self, connector_config: dict):
         self.connector_config = connector_config
-        self.connector = ElasticsearchConnector(self.config)
+        self.connector = ElasticsearchConnector(self.connector_config)
 
     async def get_documents(self):
         """Get documents from Elasticsearch (kept for backward compatibility)"""