Google Cloud Functions

An example integration where a Data Connector forwards events to a Google Cloud Function.

Overview

This example uses a Data Connector to forward the events of all devices in a project to a Google Cloud Function. When receiving the HTTPS POST request, our function will verify both the origin and content of the request using a signature secret, then decode the data.

Prerequisites

The following points are assumed.

• You have the role of Project Developer or higher in your DT Studio project.

• You are familiar with the Introduction to Data Connectors and know how to Create a Data Connector.

• You are familiar with the Cloud Functions documentation.

Google Cloud Platform

While there are many advantages to using a local environment for development, this guide will be using the browser portal to minimize setup requirements.

Create a Cloud Function

Following this guide, create a new Cloud Function with the following configurations.

Python 3.11

(1) Configuration

Basics

• Environment: 2nd gen

• Function name: As desired.

• Region: As desired.

Trigger HTTPS

• Authentication: Allow unauthenticated invocations.

Runtime, build, connections, and security settings

Add a new runtime environment variable with the following values.

• Name: DT_SIGNATURE_SECRET

• Value: A unique password. We will need it later, so write it down.

(2) Code

• Runtime: Python 3.11

• Entry point: dataconnector_endpoint

In the Source Code, edit main.py with the following snippet. The implementation is explained in detail on the Data Connector Advanced Configurations page.

main.py

import os
import hashlib

import jwt
import functions_framework

# Fetch environment variable secret.
SIGNATURE_SECRET = os.environ.get('DT_SIGNATURE_SECRET')


def verify_request(body, token):
    """
    Verifies that the request originated from DT and that the body
    hasn't been modified since it was sent. This is done by verifying
    that the checksum field of the JWT token matches the checksum of the
    request body and that the JWT is signed with the signature secret.
    """

    # Decode the token using signature secret.
    try:
        payload = jwt.decode(token, SIGNATURE_SECRET, algorithms=["HS256"])
    except Exception as err:
        print(err)
        return False

    # Verify the request body checksum.
    m = hashlib.sha1()
    m.update(body)
    checksum = m.digest().hex()
    if payload["checksum"] != checksum:
        print('Checksum Mismatch')
        return False

    return True


@functions_framework.http
def dataconnector_endpoint(request):
    # Extract the body as a byte string and the signed JWT.
    # We'll use these values to verify the request.
    payload = request.get_data()
    token = request.headers['x-dt-signature']

    # Verify request origin and content integrity.
    if not verify_request(payload, token):
        return ('Could not verify request.', 400)

    # Decode the body as JSON
    body = request.get_json(silent=True)

    # Fetch some information about the event, then print.
    # You can implement your own logic here as desired.
    event_type: str = body['event']['eventType']
    device_type: str = body['metadata']['deviceType']
    device_id: str = body['metadata']['deviceId']
    print(f'Got {event_type} event from {device_type} device {device_id}.')

    return ('OK', 200)

Edit requirements.txt with the following entries.

requirements.txt

functions-framework==3.*
pyjwt==2.7.0

Python API

(1) Configuration

Basics

• Environment: 2nd gen

• Function name: As desired.

• Region: As desired.

Trigger HTTPS

• Authentication: Allow unauthenticated invocations.

Runtime, build, connections, and security settings

Add a new runtime environment variable with the following values.

• Name: DT_SIGNATURE_SECRET

• Value: A unique password. We will need it later, so write it down.

(2) Code

• Runtime: Python 3.11

• Entry point: dataconnector_endpoint

In the Source Code, edit main.py with the following snippet. For details, read our Python API Documentation.

main.py

import os
import functions_framework

from dtintegrations import data_connector, provider


@functions_framework.http
def dataconnector_endpoint(request):
    # Validate and decode the incoming request.
    payload = data_connector.HttpPush.from_provider(
        request=request,
        provider=provider.GCLOUD,
        secret=os.getenv('DT_SIGNATURE_SECRET'),
    )

    # Print the payload data.
    print(payload)

    # If all is well, return a 200 response.
    return ('OK', 200)

Edit requirements.txt with the following entries.

requirements.txt

functions-framework==3.*
dtintegrations==0.5.1

Node.js 20

(1) Configuration

Basics

• Environment: 2nd gen

• Function name: As desired.

• Region: As desired.

Trigger HTTPS

• Authentication: Allow unauthenticated invocations.

Runtime, build, connections, and security settings

Add a new runtime environment variable with the following values.

• Name: DT_SIGNATURE_SECRET

• Value: A unique password. We will need it later, so write it down.

(2) Code

• Runtime: Node.js 20

• Entry point: dataconnectorEndpoint

In the Source Code, edit index.js with the following snippet. The implementation is explained in detail on the Data Connector Advanced Configurations page.

index.js

const crypto = require('crypto')
const jwt = require('jsonwebtoken')
const functions = require('@google-cloud/functions-framework')

// Fetch environment variables
const signatureSecret = process.env.DT_SIGNATURE_SECRET

// Verifies that the request originated from DT, and that the body
// hasn't been modified since it was sent. This is done by verifying
// that the checksum field of the JWT token matches the checksum of the
// request body, and that the JWT is signed with the signature secret.
function verifyRequest(body, token) {
    // Decode the JWT, and verify that it was 
    // signed using the signature secret.
    let decoded
    try {
        decoded = jwt.verify(token, signatureSecret)
    } catch(err) {
        console.log(err)
        return false
    }

    // Verify the request body checksum.
    let shasum = crypto.createHash('sha1')
    let checksum = shasum.update(body).digest('hex')
    if (checksum !== decoded.checksum) {
        console.log('Checksum Mismatch')
        return false
    }

    return true
}

functions.http('dataconnectorEndpoint', (req, res) => {
    // Extract the body as a string and the signed JWT.
    // We'll use these values to verify the request. 
    let payload  = JSON.stringify(req.body)
    let token = req.get('X-Dt-Signature')
    
    // Validate request origin and content integrity.
    if (verifyRequest(payload, token) === false) {
        res.sendStatus(400)
        return
    }
    
    // First, check if the event type is one of the event
    // types we're expecting. 
    // As an example, we'll check for touch events here.
    switch (req.body.event.eventType) {
        case 'touch':
            // Now that we know this is a device event, we can 
            // check for the device type and device identifier 
            // in the event metadata.
            deviceType = req.body.metadata.deviceType
            deviceId = req.body.metadata.deviceId

            console.log(`Received touch event from ${deviceType} sensor with id ${deviceId}`)
            break
        default:
            break
    }

    res.sendStatus(200);
});

Edit package.json to contain the following dependencies field.

package.json

{
  "dependencies": {
    "@google-cloud/functions-framework": "^3.0.0",
    "jsonwebtoken": "^9.0.1"
  }
}

Go 1.20

(1) Configuration

Basics

• Environment: 2nd gen

• Function name: As desired.

• Region: As desired.

Trigger HTTPS

• Authentication: Allow unauthenticated invocations.

Runtime, build, connections, and security settings

Add a new runtime environment variable with the following values.

• Name: DT_SIGNATURE_SECRET

• Value: A unique password. We will need it later, so write it down.

(2) Code

• Runtime: Go 1.20

• Entry point: DataconnectorEndpoint

In the Source Code, edit function.go with the following snippet. The implementation is explained in detail on the Data Connector Advanced Configurations page.

function.go

package dconendpoint

import (
    "crypto/sha1"
    "encoding/hex"
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net/http"
    "os"

    "github.com/GoogleCloudPlatform/functions-framework-go/functions"
    jwt "github.com/golang-jwt/jwt/v5"
)

func init() {
    functions.HTTP("DataconnectorEndpoint", DataconnectorEndpoint)
}

// Environment variables.
var signatureSecret = os.Getenv("DT_SIGNATURE_SECRET")

// verifyRequest verifies that the request originated from DT and that the
// body hasn't been modified since it was sent. This is done by verifying
// that the checksum field of the JWT token matches the checksum of the
// request body, and that the JWT is signed with the signature secret.
func verifyRequest(bodyBytes []byte, tokenString string) error {
    // Decode the JWT, and verify that it was signed using the signature secret.
    // Also verifies the algorithm used to sign the JWT.
    token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
        // Return out signature secret to verify that it was used to sign the JWT.
        return []byte(signatureSecret), nil
    }, jwt.WithValidMethods([]string{"HS256"}))
    if err != nil {
        return err
    }

    // Verify the request body checksum.
    sha1Bytes := sha1.Sum(bodyBytes)
    sha1String := hex.EncodeToString(sha1Bytes[:])
    claims := token.Claims.(jwt.MapClaims)
    if sha1String != claims["checksum"] {
        return fmt.Errorf("Checksum mismatch.")
    }

    return nil
}

// DataConnectorEndpoint receives, validates, and returns a response for the forwarded event.
func DataconnectorEndpoint(w http.ResponseWriter, r *http.Request) {
    // Extract the body and the signed JWT.
    // We'll use these values to verify the request.
    tokenString := r.Header.Get("x-dt-signature")
    bodyBytes, err := ioutil.ReadAll(r.Body)
    if err != nil {
        fmt.Println(err)
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // Validate request origin and content integrity.
    if err := verifyRequest(bodyBytes, tokenString); err != nil {
        fmt.Println(err)
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // We now know the request came from DT Cloud and the integrity
    // of the body has been verified. We can now handle the event safely.
    if err := handleEvent(bodyBytes); err != nil {
        fmt.Println(err)
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    // Respond with a 200 status code to ack the event,
    w.WriteHeader(http.StatusOK)
    _, _ = w.Write([]byte("OK"))
}

func handleEvent(payload []byte) error {
    // The structure of the events we'll receive from a Data Connector.
    type Event struct {
        Event struct {
            EventId   string          `json:"eventId"`
            EventType string          `json:"eventType"`
            Data      json.RawMessage `json:"data"`
            Timestamp string          `json:"timestamp"`
        } `json:"event"`
        Labels   map[string]string `json:"labels"`
        Metadata map[string]string `json:"metadata"`
    }

    // The structure of the `Event.Data` field for a touch event.
    // We'll be using touch events for this example.
    type TouchData struct {
        Touch struct {
            Timestamp string `json:"updateTime"`
        } `json:"touch"`
    }

    // Decode the event
    var event Event
    if err := json.Unmarshal(payload, &event); err != nil {
        return err
    }

    // First, check if the event type is one of the
    // event-types we're expecting.
    // As an example, we'll check for touch events here.
    switch event.Event.EventType {
    case "touch":
        // Now that we know this is a touch event, we can decode
        // the `Event.Data` field.
        var touchData TouchData
        if err := json.Unmarshal(event.Event.Data, &touchData); err != nil {
            return err
        }

        // Also, since we now know this is a device event, we can
        // check for the device type and device identifier
        // in the event metadata.
        deviceType := event.Metadata["deviceType"]
        deviceId := event.Metadata["deviceId"]
        timestamp := touchData.Touch.Timestamp

        fmt.Printf("Received touch event at %s from %s sensor with id %s\n",
            timestamp,
            deviceType,
            deviceId,
        )
    }

    return nil
}

Replace the content of go.mod with the following snippet.

module example.com/gcf

require (
	github.com/GoogleCloudPlatform/functions-framework-go v1.5.2
	github.com/golang-jwt/jwt/v5 v5.0.0
)

When configured, deploy your function.

Post Deployment

Your function is now ready to receive requests, but we need to know the target URL. In your function, locate the TRIGGER tab and copy the Trigger URL. Save this for later.

Create a Data Connector

To continuously forward the data to our newly created Cloud Function, a Data Connector with almost all default settings is sufficient. If you are unfamiliar with how Data Connectors can be created, refer to our Creating a Data Connector guide using the following configurations.

• Endpoint URL: The Trigger URL we found in the previous step.

• Signature Secret: The value of the DT_SIGNATURE_SECRET environment variable.

Depending on your integration, it can also be smart to disable the event types you are not interested in. For instance, the networkStatus event is sent every periodic heartbeat and will by default be forwarded by the Data Connector if not explicitly unticked.

Test the Integration

If the integration was correctly implemented, the Success counter for your Data Connector should increment for each new event forwarded. This happens each periodic heartbeat or by touching a sensor to force a new event.

If instead the Error counter increments, a response containing a non-2xx status code is returned.

• Verify that the Data Connector endpoint URL is correct.

• Google provides a host of tools that monitor Cloud Functions. Check the logs for any tracebacks that could explain why an error is returned.

Next steps

Your sensor data is now in the Google Cloud environment, and you can start using it in their various services. Fortunately, Google has some well-documented guides to get you started.

PostgreSQL Database

A database should be tailored to each specific use case. However, if you're uncertain, PostgreSQL (Postgres) is a good place to get started. The following guides will show you how to create a new Postgres database, then connect your Cloud Function to execute queries.

• Quickstart for Cloud SQL for PostgreSQL.

• Connecting from Cloud Functions to Cloud SQL.

DataStudio

Once your Database has been set up, the following guide shows you how to connect it to DataStudio for continuous data visualization and analytics.

• Connect Data Studio to Cloud SQL.