Working, simplified template

Author: PetrHeinz

Dockerfile

@@ -11,4 +11,4 @@ COPY pipeline.py .
 ENV FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE="${WORKDIR}/requirements.txt"
 ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/pipeline.py"
 
-RUN pip install -U -r ./requirements.txt 
+RUN pip install -U -r ./requirements.txt 

README.md

@@ -1,31 +1,13 @@
-# GCP Dataflow PubSub to Better Stack
+# Google Cloud Pub/Sub to Better Stack
 
-A Google Cloud Dataflow Flex template that reads messages from PubSub and sends them to Better Stack.
-
-## Overview
-
-This template provides a scalable solution for streaming data from Google Cloud PubSub to Better Stack. It uses Apache Beam's Python SDK and can be deployed as a Dataflow Flex template.
-
-## Prerequisites
-
-- Google Cloud Platform account with Dataflow and PubSub enabled
-- Better Stack account with a source token
-- Docker installed (for building the template)
-- Google Cloud SDK installed
-
-## Environment Variables
-
-The template requires two environment variables:
-
-- `BETTER_STACK_SOURCE_TOKEN`: Your Better Stack source token
-- `BETTER_STACK_INGESTING_HOST`: The Better Stack ingest host URL
+A Dataflow Flex template that reads messages from Pub/Sub and sends them to Better Stack Telemetry.
 
 ## Building and Deploying the Template
 
 1. Clone this repository:
 ```bash
-git clone https://github.com/your-org/gcp-dataflow-pubsub-to-telemetry.git
-cd gcp-dataflow-pubsub-to-telemetry
+git clone https://github.com/your-org/gcp-dataflow-pubsub-to-betterstack.git
+cd gcp-dataflow-pubsub-to-betterstack
 ```
 
 2. Choose Google Cloud Platform project to use
@@ -37,75 +19,43 @@ gcloud projects list
 gcloud config set project PROJECT_ID
 ```
 
-3. Build the Docker image:
+3. Choose Google Cloud Platform region to use
 ```bash
-docker build -t gcr.io/$(gcloud config get-value project)/pubsub-to-betterstack .
+# See currently selected region
+gcloud config get-value compute/region
+# You can switch to a different region using
+gcloud app regions list
+gcloud config set compute/region PROJECT_ID
 ```
 
-4. Push the image to Google Container Registry:
-```bash
-gcloud auth configure-docker
-docker push gcr.io/$(gcloud config get-value project)/pubsub-to-betterstack
-```
-
-5. Create a Cloud Storage bucket for the template (if you don't have one):
+4. Create a Cloud Storage bucket for the template (if you don't have one):
 ```bash
 BUCKET_NAME="dataflow-templates-$(gcloud config get-value project)"
-gsutil mb -l us-central1 gs://${BUCKET_NAME}
+gsutil mb -l $(gcloud config get-value compute/region) gs://${BUCKET_NAME}
 ```
 
-6. Update the template specification with your project ID:
+5. Set parameters based on your Google Cloud Pub/Sub Subscription and Better Stack Telemetry source
 ```bash
-sed -i "s/PROJECT_ID/$(gcloud config get-value project)/g" pubsub-to-betterstack.json
+INPUT_SUBSCRIPTION=projects/$(gcloud config get-value project)/subscriptions/<your-pubsub-subscription-name>
+SOURCE_TOKEN=<your-better-stack-source-token>
+INGESTING_HOST=<your-better-stack-ingesting-host>
 ```
 
-7. Upload the template specification to Cloud Storage:
-```bash
-gsutil cp pubsub-to-betterstack.json gs://${BUCKET_NAME}/templates/
-```
-
-8. Deploy the template using gcloud CLI:
+6. Build, deploy and run the template
 ```bash
+gcloud builds submit --tag "gcr.io/$(gcloud config get-value project)/pubsub-to-betterstack" .
+gcloud dataflow flex-template build gs://$BUCKET_NAME/pubsub-to-betterstack.json \
+    --image "gcr.io/$(gcloud config get-value project)/pubsub-to-betterstack" \
+    --sdk-language "PYTHON" \
+    --metadata-file "metadata.json"
 gcloud dataflow flex-template run "pubsub-to-betterstack-$(date +%Y%m%d-%H%M%S)" \
-    --template-file-gcs-location=gs://${BUCKET_NAME}/templates/pubsub-to-betterstack.json \
-    --parameters input_subscription=projects/$(gcloud config get-value project)/subscriptions/YOUR_SUBSCRIPTION \
-    --parameters better_stack_source_token=YOUR_SOURCE_TOKEN \
-    --parameters better_stack_ingesting_host=YOUR_INGESTING_HOST \
-    --region=$(gcloud config get-value compute/region) \
-    --additional-experiments=use_runner_v2
-```
-
-### Using Google Cloud Console
-
-1. Go to the Dataflow section in the Google Cloud Console
-2. Click "Create Job from Template"
-3. Select "Custom Template"
-4. Enter the path to your template in Cloud Storage: `gs://${BUCKET_NAME}/templates/pubsub-to-betterstack.json`
-5. Fill in the required parameters:
-   - `input_subscription`: Your PubSub subscription to read from
-   - `better_stack_source_token`: Your Better Stack source token
-   - `better_stack_ingesting_host`: The Better Stack ingest host URL
-6. Click "Run Job"
-
-## Message Format
-
-The template expects messages in JSON format. Each message will be sent to Better Stack as-is. For example:
-
-```json
-{
-  "message": "Hello from PubSub",
-  "timestamp": "2024-02-11T12:00:00Z",
-  "severity": "INFO"
-}
+    --template-file-gcs-location=gs://$BUCKET_NAME/pubsub-to-betterstack.json \
+    --parameters input_subscription=$INPUT_SUBSCRIPTION \
+    --parameters better_stack_source_token=$SOURCE_TOKEN \
+    --parameters better_stack_ingesting_host=$INGESTING_HOST \
+    --region=$(gcloud config get-value compute/region)
 ```
 
-## Error Handling
-
-The template includes error handling that:
-- Logs errors but continues processing
-- Retries failed requests to Better Stack
-- Maintains message ordering
-
 ## License
 
 ISC License. See [LICENSE.md](LICENSE.md) for details.

pubsub-to-betterstack.json renamed to metadata.json

@@ -1,29 +1,25 @@
 {
   "name": "pubsub-to-betterstack",
-  "description": "Dataflow template to stream data from PubSub to Better Stack",
+  "description": "Dataflow template to stream data from Pub/Sub to Better Stack",
   "parameters": [
     {
       "name": "input_subscription",
-      "label": "Input PubSub subscription",
-      "helpText": "PubSub subscription to read from",
+      "label": "Input Pub/Sub subscription",
+      "helpText": "The name of the Pub/Sub subscription to read from",
       "isOptional": false,
       "regexes": ["^projects/[^/]+/subscriptions/[^/]+$"]
     },
     {
       "name": "better_stack_source_token",
       "label": "Better Stack Source Token",
-      "helpText": "Your Better Stack source token",
+      "helpText": "The source token of your telemetry source in Better Stack",
       "isOptional": false
     },
     {
       "name": "better_stack_ingesting_host",
       "label": "Better Stack Ingesting Host",
-      "helpText": "The Better Stack ingest host URL",
+      "helpText": "The ingesting host of your telemetry source in Better Stack",
       "isOptional": false
     }
-  ],
-  "image": "gcr.io/PROJECT_ID/pubsub-to-betterstack:latest",
-  "sdkInfo": {
-    "language": "PYTHON"
-  }
-} 
+  ]
+}

pipeline.py

@@ -7,24 +7,28 @@
 from typing import Dict, Any
 
 class PubSubToBetterStack(beam.DoFn):
-    def __init__(self, source_token: str, ingest_host: str):
+    def __init__(self, source_token: str, ingesting_host: str):
         self.source_token = source_token
-        self.ingest_host = ingest_host
+        self.ingesting_url = ingesting_host if '://' in ingesting_host else f'https://{ingesting_host}'
         self.headers = {
             'Authorization': f'Bearer {source_token}',
             'Content-Type': 'application/json'
         }
 
     def process(self, element: bytes) -> None:
         try:
-            # Parse the PubSub message
-            message = json.loads(element.decode('utf-8'))
-            
+            # Parse the Pub/Sub data
+            data = json.loads(element.decode('utf-8'))
+
+            # Rename timestamp key to dt to be understood by Better Stack
+            if 'timestamp' in data:
+                data['dt'] = data.pop('timestamp')
+
             # Send to Better Stack
             response = requests.post(
-                self.ingest_host,
+                self.ingesting_url,
                 headers=self.headers,
-                json=message
+                json=data
             )
 
             if response.status_code != 202:
@@ -39,17 +43,17 @@ def run(argv=None):
     parser.add_argument(
         '--input_subscription',
         required=True,
-        help='Input PubSub subscription to read from'
+        help='The name of the Pub/Sub subscription to read from'
     )
     parser.add_argument(
         '--better_stack_source_token',
         required=True,
-        help='Better Stack Telemetry source token'
+        help='The source token of your telemetry source in Better Stack'
     )
     parser.add_argument(
         '--better_stack_ingesting_host',
         required=True,
-        help='Better Stack Telemetry source ingesting host'
+        help='The ingesting host of your telemetry source in Better Stack'
     )
     known_args, pipeline_args = parser.parse_known_args(argv)
 
@@ -61,7 +65,7 @@ def run(argv=None):
     with beam.Pipeline(options=pipeline_options) as p:
         (
             p
-            | 'Read from PubSub' >> beam.io.ReadFromPubSub(
+            | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(
                 subscription=known_args.input_subscription
             )
             | 'Send to Better Stack' >> beam.ParDo(

setup.py

@@ -8,4 +8,4 @@
         'requests>=2.31.0',
     ],
     packages=setuptools.find_packages(),
-) 
+)