DEV Community: Mohamed Rasvi

Deploy Kafka connector on GKE cluster

Mohamed Rasvi — Mon, 14 Oct 2024 17:12:27 +0000

Here are some scenarios in which you might use the Pub/Sub Group Kafka Connector:

You are migrating a Kafka-based architecture to Google Cloud.
You have a frontend system that stores events in Kafka outside of Google Cloud, but you also use Google Cloud to run some of your backend services, which need to receive the Kafka events.
You collect logs from an on-premises Kafka solution and send them to Google Cloud for data analytics.
You have a frontend system that uses Google Cloud, but you also store data on-premises using Kafka.

As with Kafka, you can use Pub/Sub to communicate between components in your cloud architecture.

The Pub/Sub Group Kafka Connector allows you to integrate these two systems. The following connectors are packaged in the Connector JAR:

The sink connector reads records from one or more Kafka topics and publishes them to Pub/Sub.
The source connector reads messages from a Pub/Sub topic and publishes them to Kafka.

This document we are going to walk through how we can set up sink connector
More information

This section walks you through the following tasks:

Configure the Pub/Sub Group Kafka Connector.
Send events from Kafka to Pub/Sub.

Authenticate
The Pub/Sub Group Kafka Connector must authenticate with Pub/Sub in order to send Pub/Sub messages. To set up authentication, perform the following steps:

Grant roles to your Google Service Account, IAM roles: roles/pubsub.admin

git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
cd java-pubsub-group-kafka-connector

Download the connector JAR

Download JAR

cp config/* [path to Kafka installation]/config/

Update your Kafka Connect configuration

Navigate to your Kafka directory.
Open the file named config/connect-standalone.properties in a text editor.
If the plugin.path property is commented out, uncomment it.
Update the plugin.path property to include the path to the connector JAR. Example:

plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar

Set the offset.storage.file.filename property to a local file name. In standalone mode, Kafka uses this file to store offset data.
Example:

offset.storage.file.filename=/tmp/connect.offsets

Forward events from Kafka to Pub/Sub

Open the file /config/cps-sink-connector.properties in a text editor. Add values for the following properties, which are marked "TODO" in the comments:
topics=KAFKA_TOPICS
cps.project=PROJECT_ID
cps.topic=PUBSUB_TOPIC
gcp.credentials.file.path=PATH
gcp.credentials.json = JSON_FILE
Replace the following:

KAFKA_TOPICS: A comma-separated list of Kafka topics to read from.
PROJECT_ID: The Google Cloud project that contains your Pub/Sub topic.
SUB_TOPIC: The Pub/Sub topic to receive the messages from Kafka
PATH String Optional. The path to a file that stores Google Cloud credentials for authenticating Pub/Sub (DOC says pub/sub lite but I guess should work on pub/sub also)
JSON_FILE Optional. A JSON blob that contains Google Cloud for authenticating Pub/Sub (DOC says pub/sub lite but I guess should work on pub/sub also)

More info about settings
File: cps-sink-connector.properties

# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# Unique name for the Pub/Sub sink connector.
name=CPSSinkConnector
# Tha Java class for the Pub/Sub sink connector.
connector.class=com.google.pubsub.kafka.sink.CloudPubSubSinkConnector
# The maximum number of tasks that should be created for this connector.
tasks.max=10
# Set the key converter for the Pub/Sub sink connector.
key.converter=org.apache.kafka.connect.storage.StringConverter
# Set the value converter for the Pub/Sub sink connector.
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
# A comma-seperated list of Kafka topics to use as input for the connector.
# TODO (developer): update to your Kafka topic name(s).
topics=scotia-wealth
cps.project=PROJECT-ID
# TODO (developer): update to your Pub/Sub topic ID, e.g.
# where data should be written.
cps.topic=kafka-topic-consumer
# Optional. A JSON file path and JSON blob that contains Google Cloud for authenticating Pub/Sub Lite.
gcp.credentials.file.path=PATH
gcp.credentials.json=JSON_FILE

From the Kafka directory, run the following command:

bin/connect-standalone.sh \
  config/connect-standalone.properties \
  config/cps-sink-connector.properties

Follow the steps in the Apache Kafka quickstart to write some events to your Kafka topic.
Use the gcloud CLI to read the events from Pub/Sub.
gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack

more info about kafka
GCP repo :https://github.com/googleapis/java-pubsub-group-kafka-connector

Ok, we have tested locally. Now it is a time to containerize and deploy the Kafka connector on GKE

Deploy kafka connector on GKE as following

Build the image from the following based image

[https://github.com/bitnami/containers/tree/main/bitnami/kafka](https://github.com/bitnami/containers/tree/main/bitnami/kafka)

Built a custom Image with connector :- Pub/Sub Group Kafka Connector .Jar
example dockerfile :

FROM bitnami/kafka:3.4.0-debian-11-r15 AS build-stage
COPY YOUR_pubsubKafkaConnector_FOLDER /opt/bitnami/kafka/
COPY setup /opt/bitnami/kafka/

Create a pubsub
Topic
Subscription
Deploy the helm chart :

Installing the Chart

Build the image from Dockerfile
## Prerequisites
https://cloud.google.com/artifact-registry/docs/docker/pushing-and-pulling

docker build -t LOCATION-docker.pkg.dev/PROJECT-ID/REPOSITORY/IMAGE:TAG .
docker push LOCATION-docker.pkg.dev/PROJECT-ID/REPOSITORY/IMAGE:TAG

## Helm installs the Kafka instance with a pub/sub connector from a custom image (a custom image that we have built previously)
helm repo add bitnami https://charts.bitnami.com/bitnami

change the docker image in kafka chart (kafka/value.yaml)

## Bitnami Kafka image version
## ref: https://hub.docker.com/r/bitnami/kafka/tags/
## @param image.registry Kafka image registry
## @param image.repository Kafka image repository
## @param image.tag Kafka image tag (immutable tags are recommended)
## @param image.digest Kafka image digest in the way sha256:aa.... Please note this parameter, if set, will override the tag
## @param image.pullPolicy Kafka image pull policy
## @param image.pullSecrets Specify docker-registry secret names as an array
## @param image.debug Specify if debug values should be set
##
image:
  registry: LOCATION-docker.pkg.dev/PROJECT-ID/REPOSITORY
  repository: IMAGE
  tag: TAG
  digest: ""

Your GKE cluster looks like After deploying the helm chart

1. Statefulset
     Kafka instance 
     Kafka zookeeper

2. K8s service 
     Exposed GKE service to publish and consume (kafka topic and messages)

impersonate GCP Service account via workload identity

Service account which has permissions to publish message (k8s
SA will impersonate GCP SA via workload identity ) more info

following example:

## Workload identity to impersonate gcp SA
gcloud iam service-accounts add-iam-policy-binding $GSA \
    --role roles/iam.workloadIdentityUser \
    --member "serviceAccount:$PROJECT_ID.svc.id.goog[kafka/kafka]"


## Annotate k8s serviceaccount kafka
kubectl annotate serviceaccount kafka \
    --namespace kafka \
    iam.gke.io/gcp-service-account=$GSA

Create the ConfigMap from previously downloaded config map files
Following the example, we have created a folder called config and saved config files

kubectl create configmap pubsub-connector --from-file=config/ -n kafka

AI powered chat agent to manage public cloud

Mohamed Rasvi — Wed, 09 Oct 2024 20:17:23 +0000

I'm thinking to launch a product, AI powered chat aget to manage public cloud infrastructure.

AI powered chat agent to manage public cloud
Join Our Waiting List!
Be the first to know when our product launches. Sign up now!
AI powered chat agent to manage public cloud

Deploying a stateless container on cloud run

Mohamed Rasvi — Mon, 07 Oct 2024 21:43:21 +0000

I will demonstrate how to deploy a simple container on cloud run.

Cloud Run is a fully managed platform that enables you to run your code directly on top of Google’s scalable infrastructure. Cloud Run is simple, automated, and designed to make you more productive.

Create a simple hello world application using fastapi library (python)
Containerize the application
Configure the workflow with GCP
Deploy the container onto cloud run service via github workflow

I followed official fastapi doc to spin up a hello world app
Create a requirements.txt file

fastapi[standard]
pydantic>=2.7.0,<3.0.0

Create an app directory and enter it
Create an empty file init.py
Create a main.py file with:

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

Create a Dockerfile

FROM python:3.9

WORKDIR /code

COPY ./requirements.txt /code/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt

COPY ./app /code/app

CMD ["fastapi", "run", "app/main.py", "--port", "80"]

GitHub Action
In order for the GitHub actions process to pick up the YAML file, there’s specific location for it to live. Each repository using actions requires a directory structure called /.github/workflows

*Configure this workflow with GCP more info *

# This workflow build and push a Docker container to Google Artifact Registry
# and deploy it on Cloud Run when a commit is pushed to the $default-branch
# branch.
#
# To configure this workflow:
#
# 1. Enable the following Google Cloud APIs:
#
#    - Artifact Registry (artifactregistry.googleapis.com)
#    - Cloud Run (run.googleapis.com)
#    - IAM Credentials API (iamcredentials.googleapis.com)
#
#    You can learn more about enabling APIs at
#    https://support.google.com/googleapi/answer/6158841.
#
# 2. Create and configure a Workload Identity Provider for GitHub:
#    https://github.com/google-github-actions/auth#preferred-direct-workload-identity-federation.
#
#    Depending on how you authenticate, you will need to grant an IAM principal
#    permissions on Google Cloud:
#
#    - Artifact Registry Administrator (roles/artifactregistry.admin)
#    - Cloud Run Developer (roles/run.developer)
#
#    You can learn more about setting IAM permissions at
#    https://cloud.google.com/iam/docs/manage-access-other-resources
#
# 3. Change the values in the "env" block to match your values.

Create a file google-cloudrun-docker.yml

name: 'Build and Deploy to Cloud Run'

on:
  push:
    branches:
      - '$default-branch'

env:
  PROJECT_ID: 'my-project' # TODO: update to your Google Cloud project ID
  REGION: 'us-central1' # TODO: update to your region
  SERVICE: 'my-service' # TODO: update to your service name
  WORKLOAD_IDENTITY_PROVIDER: 'projects/123456789/locations/global/workloadIdentityPools/my-pool/providers/my-provider' # TODO: update to your workload identity provider

jobs:
  deploy:
    runs-on: 'ubuntu-latest'

    permissions:
      contents: 'read'
      id-token: 'write'

    steps:
      - name: 'Checkout'
        uses: 'actions/checkout@692973e3d937129bcbf40652eb9f2f61becf3332' # actions/checkout@v4

      # Configure Workload Identity Federation and generate an access token.
      #
      # See https://github.com/google-github-actions/auth for more options,
      # including authenticating via a JSON credentials file.
      - id: 'auth'
        name: 'Authenticate to Google Cloud'
        uses: 'google-github-actions/auth@f112390a2df9932162083945e46d439060d66ec2' # google-github-actions/auth@v2
        with:
          workload_identity_provider: '${{ env.WORKLOAD_IDENTITY_PROVIDER }}'

      # BEGIN - Docker auth and build
      #
      # If you already have a container image, you can omit these steps.
      - name: 'Docker Auth'
        uses: 'docker/login-action@9780b0c442fbb1117ed29e0efdff1e18412f7567' # docker/login-action@v3
        with:
          username: 'oauth2accesstoken'
          password: '${{ steps.auth.outputs.auth_token }}'
          registry: '${{ env.REGION }}-docker.pkg.dev'

      - name: 'Build and Push Container'
        run: |-
          DOCKER_TAG="$${{ env.REGION }}-docker.pkg.dev/${{ env.PROJECT_ID }}/${{ env.SERVICE }}:${{ github.sha }}"
          docker build --tag "${DOCKER_TAG}" .
          docker push "${DOCKER_TAG}"
      - name: 'Deploy to Cloud Run'

        # END - Docker auth and build

        uses: 'google-github-actions/deploy-cloudrun@33553064113a37d688aa6937bacbdc481580be17' # google-github-actions/deploy-cloudrun@v2
        with:
          service: '${{ env.SERVICE }}'
          region: '${{ env.REGION }}'
          # NOTE: If using a pre-built image, update the image name below:

          image: '${{ env.REGION }}-docker.pkg.dev/${{ env.PROJECT_ID }}/${{ env.SERVICE }}:${{ github.sha }}'
      # If required, use the Cloud Run URL output in later steps
      - name: 'Show output'
        run: |2-

          echo ${{ steps.deploy.outputs.url }}

Directory Structure

You should now have a directory structure like:

├── app
│   ├── __init__.py
│   └── main.py
├── Dockerfile
└── requirements.txt
└── requirements.txt
├── .github
│   ├── workflows
         ├── google-cloudrun-docker.yml


> 1. Create a new repo in gitHUb
> 2. Push your exisisting code to new repository on default branch

Google identity Platform

Mohamed Rasvi — Fri, 04 Oct 2024 17:01:34 +0000

This post I would like to explain how google identity platform works in a simple manner.

Most of the time, the Google Identity platform works as a proxy to configure identity providers. Google identity platform is little different to Azure ID, where you can create your own SAML and OIDC identity providers.

For the user authentication In this article, we will use federated identity provider Google.

Above User flow in details

Users access the web application
Application: check whether the user is logged in or not
If you are not authenticated, the application will redirect to a federated identity provider : Google
Once the user is authenticated successfully and redirects to the /__auth/handler endpoint, this service is hosted by Google
The /__auth/handler service redirects back to the application

We need to understand the Main 5 concepts behind this google identity platform.

Authentication

Users

Admin Auth API

Multi-tenancy

Differences between Identity Platform and Firebase Authentication

I will explain the main most important concept for this post; keep the article short and simple.

Authentication
Identity Platform allows users to authenticate to your apps and services, like multi-tenant SaaS apps, mobile/web apps, games, APIs and more. Identity Platform provides secure, easy-to-use authentication if you're building a service on Google Cloud, on your own backend or on another platform.

How it works?
To sign a user into your app, you first get authentication credentials from the user. These credentials can be the user's
email address and password, a SAML assertion, or an OAuth token from a federated identity provider.
In the case of federated identity providers, the providers return those tokens to Identity Platform's authentication handler on the /__auth/handler endpoint. This service is hosted by Google, so you don't have to receive and validate the authentication artifact. After the tokens are received, our backend services will verify them and return a response to the client.
After a successful sign in, you can access the user's basic profile information, and you can control the user's access to data stored in Google Cloud or other products. You can also use the provided authentication token to verify the identity of users in your own backend services.

You can read here to understand the above all concepts here

Setting up a Google IDP on google idenity platform

Go to google idenity platform
Click Add A Provider and select Google
Enter your Google Web Client ID and Web Secret. If you don't already have an ID and secret, you can obtain one from the API's & Services page.
Configure the URI listed under Configure Google as a valid OAuth redirect URI for your Google app. If you configured a custom domain in Identity Platform, update the redirect URI in your Google app configuration to use the custom domain instead of the default domain. For example, change https://myproject.firebaseapp.com/__/auth/handler to https://auth.myownpersonaldomain.com/__/auth/handler. This is where confusion we see on the Google cloud document. Most of the time we don't read the concepts, so we assume our application must have this callback URL, /__/auth/handler, but we don't need this because /__/auth/handler is hosted in Google; it is a Google-hosted backend service that does the magic like creating JWT, etc.
Register your app's domains by clicking Add Domain under Authorized Domains. For development purposes, localhost is already enabled by default.
Click Save.

I'm not going to show the entire code, I just insert some example code snippets.

import { getAuth, signInWithRedirect, getRedirectResult, GoogleAuthProvider } from "firebase/auth"; const auth = getAuth(); signInWithRedirect(auth, provider); getRedirectResult(auth) .then((result) => { // This gives you a Google Access Token. You can use it to access Google APIs. const credential = GoogleAuthProvider.credentialFromResult(result); const token = credential.accessToken; // The signed-in user info. const user = result.user; // IdP data available using getAdditionalUserInfo(result) // ... }).catch((error) => { // Handle Errors here. const errorCode = error.code; const errorMessage = error.message; // The email of the user's account used. const email = error.customData.email; // The AuthCredential type that was used. const credential = GoogleAuthProvider.credentialFromError(error); // ... });

Signing in users with the Client SDK

Let me know if you want to dive deep into the Google identity platform.

For example, integration with gateways like Kong.