Initial commit

Author: PetrHeinz

Dockerfile

@@ -0,0 +1,14 @@
+FROM gcr.io/dataflow-templates-base/python3-template-launcher-base
+
+ARG WORKDIR=/dataflow/template
+RUN mkdir -p ${WORKDIR}
+WORKDIR ${WORKDIR}
+
+COPY requirements.txt .
+COPY setup.py .
+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 

LICENSE.md

@@ -0,0 +1,7 @@
+ISC License
+
+Copyright (c) 2025, Better Stack, Inc.
+
+Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

README.md

@@ -0,0 +1,89 @@
+# GCP Dataflow PubSub 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_INGEST_HOST`: The Better Stack ingest host URL
+
+## Building 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
+```
+
+2. Build the Docker image:
+```bash
+docker build -t gcr.io/YOUR_PROJECT/pubsub-to-betterstack .
+```
+
+3. Push the image to Google Container Registry:
+```bash
+docker push gcr.io/YOUR_PROJECT/pubsub-to-betterstack
+```
+
+## Deploying the Template
+
+You can deploy the template using the Google Cloud Console or the gcloud CLI:
+
+### Using gcloud CLI
+
+```bash
+gcloud dataflow flex-template run "pubsub-to-betterstack-$(date +%Y%m%d-%H%M%S)" \
+    --template-file-gcs-location=gs://YOUR_BUCKET/templates/pubsub-to-betterstack.json \
+    --parameters input_subscription=projects/YOUR_PROJECT/subscriptions/YOUR_SUBSCRIPTION \
+    --region=YOUR_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
+5. Fill in the required parameters:
+   - `input_subscription`: Your PubSub subscription to read from
+6. Set the environment variables:
+   - `BETTER_STACK_SOURCE_TOKEN`
+   - `BETTER_STACK_INGEST_HOST`
+7. 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"
+}
+```
+
+## 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.

pipeline.py

@@ -0,0 +1,72 @@
+import argparse
+import json
+import os
+import apache_beam as beam
+from apache_beam.options.pipeline_options import PipelineOptions
+import requests
+from typing import Dict, Any
+
+class PubSubToBetterStack(beam.DoFn):
+    def __init__(self, source_token: str, ingest_host: str):
+        self.source_token = source_token
+        self.ingest_host = ingest_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'))
+            
+            # Send to Better Stack
+            response = requests.post(
+                self.ingest_host,
+                headers=self.headers,
+                json=message
+            )
+            
+            if response.status_code != 202:
+                raise Exception(f"Failed to send to Better Stack: {response.text}")
+                
+        except Exception as e:
+            # Log the error but don't fail the pipeline
+            print(f"Error processing message: {str(e)}")
+
+def run(argv=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        '--input_subscription',
+        required=True,
+        help='Input PubSub subscription to read from'
+    )
+    known_args, pipeline_args = parser.parse_known_args(argv)
+
+    # Get Better Stack credentials from environment variables
+    source_token = os.environ.get('BETTER_STACK_SOURCE_TOKEN')
+    ingest_host = os.environ.get('BETTER_STACK_INGEST_HOST')
+
+    if not source_token or not ingest_host:
+        raise ValueError(
+            "Environment variables BETTER_STACK_SOURCE_TOKEN and BETTER_STACK_INGEST_HOST must be set"
+        )
+
+    pipeline_options = PipelineOptions(
+        pipeline_args,
+        save_main_session=True
+    )
+
+    with beam.Pipeline(options=pipeline_options) as p:
+        (
+            p
+            | 'Read from PubSub' >> beam.io.ReadFromPubSub(
+                subscription=known_args.input_subscription
+            )
+            | 'Send to Better Stack' >> beam.ParDo(
+                PubSubToBetterStack(source_token, ingest_host)
+            )
+        )
+
+if __name__ == '__main__':
+    run() 

requirements.txt

@@ -0,0 +1,2 @@
+apache-beam[gcp]>=2.50.0
+requests>=2.31.0 

setup.py

@@ -0,0 +1,11 @@
+import setuptools
+
+setuptools.setup(
+    name='pubsub-to-betterstack',
+    version='0.1.0',
+    install_requires=[
+        'apache-beam[gcp]>=2.50.0',
+        'requests>=2.31.0',
+    ],
+    packages=setuptools.find_packages(),
+)