chore: pass bootstrap.servers for local env (#61)

* chore: pass `bootstrap.servers` for local env

* improve CONTRIBUTING.md

* fix formatting

Author: dormeiri-port

CONTRIBUTING.md

@@ -37,24 +37,78 @@ To submit code contributions:
 
 ## Debugging
 
-### Running Against Local Port Instance
+### Set Environment Variables
 
-When debugging the Port Execution Agent locally against a local instance of Port, follow these steps:
+Create or update your `.env` file to include the following environment variables (replace with your actual values):
 
-1. **Set Up Local Environment:**
-   - Ensure you have a local instance of Port running. This will act as your development and testing environment.
+```env
+STREAMER_NAME=KAFKA
+KAFKA_CONSUMER_GROUP_ID=my_consumer_group
 
-2. **Configure Environment Variables:**
-   - Create or update your `.env` file to include the following environment variable:
-     ```env
-     USING_LOCAL_PORT_INSTANCE=True
-     ```
+PORT_ORG_ID=your_organization_id_here
+PORT_CLIENT_ID=your_client_id_here
+PORT_CLIENT_SECRET=your_client_secret_here
+PORT_API_BASE_URL=your_api_base_url_here
+ ```
 
-3. **Kafka Authentication:**
-   - When `USING_LOCAL_PORT_INSTANCE` is set to `True`, the execution agent will not attempt to pull your local organization's kafka credentials.
+### Run with Python
 
-4. **Running the Agent Locally:**
-   - Start the execution agent as you normally would.
+**Prequisites:**
+
+- Python 3.11
+- [Poetry](https://python-poetry.org/docs/#installing-manually)
+
+**Install Dependencies:**
+
+It is recommended to install the dependencies inside a virtual environment:
+
+```bash
+VENV_PATH=".venv"
+python3.11 -m venv $VENV_PATH
+$VENV_PATH/bin/pip install -U pip setuptools
+$VENV_PATH/bin/pip install poetry
+source $VENV_PATH/bin/activate
+
+poetry install
+```
+
+**Run the Agent:**
+
+Activate the virtual environment if not already activated:
+
+```bash
+source .venv/bin/activate
+```
+
+The execution must be from the `app` directory:
+
+```bash
+cd app
+```
+
+To start the Port Execution Agent, run the following command from the `app` directory:
+
+```bash
+python main.py
+```
+
+### Run with Docker
+
+Before running the container, ensure you have built the Docker image:
+
+```bash
+docker build -t port-execution-agent .
+```
+
+Run the Port Execution Agent container with the following command:
+
+```bash
+docker run \
+  --name port-execution-agent \
+  --env-file .env \
+  --network host \
+  port-execution-agent
+```
 
 ### General Troubleshooting (Optional)
 

app/consumers/kafka_consumer.py

@@ -34,7 +34,10 @@ def __init__(
                 "auto.offset.reset": settings.KAFKA_CONSUMER_AUTO_OFFSET_RESET,
                 "enable.auto.commit": "false",
             }
-            if not settings.USING_LOCAL_PORT_INSTANCE:
+            if settings.USING_LOCAL_PORT_INSTANCE:
+                logger.info("Using local Port instance for Kafka credentials")
+                conf["bootstrap.servers"] = settings.KAFKA_CONSUMER_BOOTSTRAP_SERVERS
+            else:
                 logger.info("Getting Kafka credentials")
                 brokers, username, password = get_kafka_credentials()
                 conf["sasl.username"] = username

app/core/config.py

@@ -2,15 +2,8 @@
 from typing import Any, Optional
 
 from dotenv import find_dotenv
-from pydantic import (
-    AnyHttpUrl,
-    BaseModel,
-    BaseSettings,
-    Field,
-    parse_file_as,
-    parse_obj_as,
-    validator,
-)
+from pydantic import (AnyHttpUrl, BaseModel, BaseSettings, Field,
+                      parse_file_as, parse_obj_as, validator)
 
 
 class ActionReport(BaseModel):
@@ -47,11 +40,32 @@ class Settings(BaseSettings):
     KAFKA_CONSUMER_SESSION_TIMEOUT_MS: int = 45000
     KAFKA_CONSUMER_AUTO_OFFSET_RESET: str = "earliest"
     KAFKA_CONSUMER_GROUP_ID: str = ""
+    KAFKA_CONSUMER_BOOTSTRAP_SERVERS: str = ""
 
     KAFKA_RUNS_TOPIC: str = ""
 
     CONTROL_THE_PAYLOAD_CONFIG_PATH: Path = Path("./control_the_payload_config.json")
 
+    @validator("KAFKA_CONSUMER_BOOTSTRAP_SERVERS", always=True)
+    def set_kafka_consumer_bootstrap_servers(
+        cls, v: Optional[str], values: dict
+    ) -> str:
+        using_local = values.get("USING_LOCAL_PORT_INSTANCE")
+        if isinstance(v, str) and v:
+            if using_local is False:
+                raise ValueError(
+                    "KAFKA_CONSUMER_BOOTSTRAP_SERVERS works only with "
+                    "USING_LOCAL_PORT_INSTANCE=True"
+                )
+
+            return v
+
+        elif using_local is True:
+            raise ValueError(
+                "KAFKA_CONSUMER_BOOTSTRAP_SERVERS must be set when "
+                "USING_LOCAL_PORT_INSTANCE=True"
+            )
+
     @validator("KAFKA_RUNS_TOPIC", always=True)
     def set_kafka_runs_topic(cls, v: Optional[str], values: dict) -> str:
         if isinstance(v, str) and v:

app/invokers/webhook_invoker.py

@@ -9,17 +9,13 @@
 from core.consts import consts
 from flatten_dict import flatten, unflatten
 from invokers.base_invoker import BaseInvoker
-from port_client import report_run_response, report_run_status, run_logger_factory
+from port_client import (report_run_response, report_run_status,
+                         run_logger_factory)
 from pydantic import BaseModel, Field
 from requests import Response
-from utils import (
-    decrypt_payload_fields,
-    get_invocation_method_object,
-    get_response_body,
-    log_by_detail_level,
-    response_to_dict,
-    sign_sha_256,
-)
+from utils import (decrypt_payload_fields, get_invocation_method_object,
+                   get_response_body, log_by_detail_level, response_to_dict,
+                   sign_sha_256)
 
 logging.basicConfig(level=settings.LOG_LEVEL)
 logger = logging.getLogger(__name__)