Merge pull request #9 from port-labs/PORT-5259-port-agent-report-run-…

…back-to-port

PORT-5259-port-agent-report-run-back-to-port

README.md

@@ -38,7 +38,13 @@ Here is the mapping file schema:
       "method": JQ, # Optional. default is POST. Should return one of the following string values POST / PUT / DELETE / GET 
       "headers": dict[str, JQ], # Optional. default is {}
       "body": ".body", # Optional. default is the whole payload incoming from Port.
-      "query": dict[str, JQ] # Optional. default is {}
+      "query": dict[str, JQ] # Optional. default is {},
+      "report" { # Optional. Used to report the run status back to Port right after the request is sent to the 3rd party application
+        "status": JQ, # Optional. Should return the wanted runs status
+        "link": JQ, # Optional. Should return the wanted link or a list of links
+        "summary": JQ, # Optional. Should return the wanted summary
+        "externalRunId": JQ # Optional. Should return the wanted external run id
+      }
   }
 ]
 ```
@@ -128,6 +134,19 @@ Here is the mapping file schema:
 
 </details>
 
+
+### The report mapping
+
+After the request is sent to the 3rd party application, the Port agent can report the run status back to Port.
+The report mapping is used to construct the report that will be sent to Port.
+
+The report mapping can use the following fields:
+
+`.body` - The incoming message as mentioned [Above](#the-incoming-message-to-base-your-mapping-on)
+`.request` - The request that was calculated using the control the payload mapping and sent to the 3rd party application 
+`.response` - The response that was received from the 3rd party application
+
+
 ### Examples
 
 #### Terraform Cloud
@@ -147,10 +166,20 @@ Create the following blueprint, action and mapping to trigger a Terraform Cloud
       "workspace_id": {
         "title": "Workspace Id",
         "type": "string"
+      },
+      "organization_name": {
+        "title": "Organization Name",
+        "type": "string"
+      },
+      "workspace_name": {
+        "title": "Workspace Name",
+        "type": "string"
       }
     },
     "required": [
-      "workspace_id"
+      "workspace_id",
+      "organization_name",
+      "workspace_name"
     ]
   },
   "mirrorProperties": {},
@@ -192,7 +221,8 @@ Create the following blueprint, action and mapping to trigger a Terraform Cloud
 <summary>Mapping - (Should be saved as `invocations.json`)</summary>
 
 ```json
-[{
+[
+  {
     "enabled": ".action == \"trigger_tf_run\"",
     "headers": {
       "Authorization": "\"Bearer \" + env.TF_TOKEN",
@@ -215,8 +245,14 @@ Create the following blueprint, action and mapping to trigger a Terraform Cloud
           }
         }
       }
+    },
+    "report": {
+      "status": "if .response.statusCode == 201 then \"SUCCESS\" else \"FAILURE\" end",
+      "link": "\"https://app.terraform.io/app/\" + .body.payload.entity.properties.organization_name + \"/workspaces/\" + .body.payload.entity.properties.workspace_name + \"/runs/\" + .response.json.data.id",
+      "externalRunId": ".response.json.data.id"
     }
-  }]
+  }
+]
 ```
 </details>
 
@@ -229,10 +265,10 @@ helm repo update
 
 helm install my-port-agent port-labs/port-agent \
     --create-namespace --namespace port-agent \
+    --set env.secret.PORT_CLIENT_ID=YOUR_PORT_CLIENT_ID \
+    --set env.secret.PORT_CLIENT_SECRET=YOUR_PORT_CLIENT_SECRET \
     --set env.normal.PORT_ORG_ID=YOUR_ORG_ID \
     --set env.normal.KAFKA_CONSUMER_GROUP_ID=YOUR_KAFKA_CONSUMER_GROUP \
-    --set env.secret.KAFKA_CONSUMER_USERNAME=YOUR_KAFKA_USERNAME \
-    --set env.secret.KAFKA_CONSUMER_PASSWORD=YOUR_KAFKA_PASSWORD \
     --set env.normal.KAFKA_CONSUMER_BROKERS=PORT_KAFKA_BROKERS \
     --set env.normal.STREAMER_NAME=KAFKA \
     --set env.normal.KAFKA_CONSUMER_AUTHENTICATION_MECHANISM=SCRAM-SHA-512 \
@@ -329,10 +365,10 @@ helm repo update
 
 helm install my-port-agent port-labs/port-agent \
     --create-namespace --namespace port-agent \
+    --set env.secret.PORT_CLIENT_ID=YOUR_PORT_CLIENT_ID \
+    --set env.secret.PORT_CLIENT_SECRET=YOUR_PORT_CLIENT_SECRET \
     --set env.normal.PORT_ORG_ID=YOUR_ORG_ID \
     --set env.normal.KAFKA_CONSUMER_GROUP_ID=YOUR_KAFKA_CONSUMER_GROUP \
-    --set env.secret.KAFKA_CONSUMER_USERNAME=YOUR_KAFKA_USERNAME \
-    --set env.secret.KAFKA_CONSUMER_PASSWORD=YOUR_KAFKA_PASSWORD
     --set env.normal.KAFKA_CONSUMER_BROKERS=PORT_KAFKA_BROKERS \
     --set env.normal.STREAMER_NAME=KAFKA \
     --set env.normal.KAFKA_CONSUMER_AUTHENTICATION_MECHANISM=SCRAM-SHA-512 \

app/consumers/kafka_consumer.py

@@ -6,6 +6,7 @@
 from consumers.base_consumer import BaseConsumer
 from core.config import settings
 from core.consts import consts
+from port_client import get_kafka_credentials
 
 logging.basicConfig(level=settings.LOG_LEVEL)
 logger = logging.getLogger(__name__)
@@ -24,13 +25,15 @@ def __init__(
         if consumer:
             self.consumer = consumer
         else:
+            logger.info("Getting Kafka credentials")
+            username, password = get_kafka_credentials()
             conf = {
                 "bootstrap.servers": settings.KAFKA_CONSUMER_BROKERS,
                 "client.id": consts.KAFKA_CONSUMER_CLIENT_ID,
                 "security.protocol": settings.KAFKA_CONSUMER_SECURITY_PROTOCOL,
                 "sasl.mechanism": settings.KAFKA_CONSUMER_AUTHENTICATION_MECHANISM,
-                "sasl.username": settings.KAFKA_CONSUMER_USERNAME,
-                "sasl.password": settings.KAFKA_CONSUMER_PASSWORD,
+                "sasl.username": username,
+                "sasl.password": password,
                 "group.id": settings.KAFKA_CONSUMER_GROUP_ID,
                 "session.timeout.ms": settings.KAFKA_CONSUMER_SESSION_TIMEOUT_MS,
                 "auto.offset.reset": settings.KAFKA_CONSUMER_AUTO_OFFSET_RESET,

app/core/config.py

@@ -1,7 +1,22 @@
 from pathlib import Path
 from typing import Any, Literal, Optional
 
-from pydantic import BaseModel, BaseSettings, parse_file_as, validator
+from pydantic import (
+    AnyHttpUrl,
+    BaseModel,
+    BaseSettings,
+    Field,
+    parse_file_as,
+    parse_obj_as,
+    validator,
+)
+
+
+class ActionReport(BaseModel):
+    status: str | None = None
+    link: str | None = None
+    summary: str | None = None
+    external_run_id: str | None = Field(None, alias="externalRunId")
 
 
 class Mapping(BaseModel):
@@ -11,6 +26,7 @@ class Mapping(BaseModel):
     body: dict[str, Any] | str | None = None
     headers: dict[str, str] | str | None = None
     query: dict[str, str] | str | None = None
+    report: ActionReport | None = None
 
 
 class Settings(BaseSettings):
@@ -19,11 +35,12 @@ class Settings(BaseSettings):
     STREAMER_NAME: str
 
     PORT_ORG_ID: str
+    PORT_API_BASE_URL: AnyHttpUrl = parse_obj_as(AnyHttpUrl, "https://api.getport.io")
+    PORT_CLIENT_ID: str
+    PORT_CLIENT_SECRET: str
     KAFKA_CONSUMER_BROKERS: str = "localhost:9092"
     KAFKA_CONSUMER_SECURITY_PROTOCOL: str = "plaintext"
     KAFKA_CONSUMER_AUTHENTICATION_MECHANISM: str = "none"
-    KAFKA_CONSUMER_USERNAME: str = "local"
-    KAFKA_CONSUMER_PASSWORD: str = ""
     KAFKA_CONSUMER_SESSION_TIMEOUT_MS: int = 45000
     KAFKA_CONSUMER_AUTO_OFFSET_RESET: str = "earliest"
     KAFKA_CONSUMER_GROUP_ID: str = ""