Search for a command to run...
A focused boutique consulting partner to boot your current and future digital DevSecOps and Cloud Platform needs.
At BootLabs we are obsessed about building future-ready Intelligent Enterprise. We solve cloud automation problems with cutting-edge technologies while making sure the TCO is minimal. We enable organisational transformation to support the rapid application development and consistent releases.
No comments yet. Be the first to comment.
A Proactive Approach to Software Security in the Digital Transformation
Gatekeeper is a general-purpose policy engine based on Open Policy Agent (OPA). It allows you to define and enforce custom policies across various Kubernetes resources. Kyverno is specifically designed for Kubernetes policy management. It focuses o...
Connecting to an AWS private EC2 instance using OpenSSH requires a few prerequisites and steps. Prerequisites: Ensure you have OpenSSH installed. Make sure you have AWS CLI version 2.12.6 or above installed. Set up an EC2 instance connect endpoint...
HANDLING SECRETS AND PARAMETERS ON AWS EKS Security best practices need the protection of personal data (e.g. passwords, tokens, API Keys). These details are often saved in AWS Secrets Manager or AWS Systems Manager Parameter Store when utilising AWS...
Velero Velero is an open-source tool for Kubernetes, to safely backup and restores the data in a cluster and disaster recovery of the cluster. With the help of velero, we can store the backup of EKS and volumes will store in a bucket and disk (in AWS...
Gin is the web framework written in Go(GoLang). Gin is a high-performance micro-framework that can be used to build web applications. It allows you to write middleware that can be plugged into one or more request handlers or groups of request handlers.
Goals
By the end of this tutorial, you will:
Prerequisites
For this tutorial, you will need GoLang, Temporal, docker, and postman installed on your machine.
Note: If you don’t have postman, you can use any other tool that you would use to test API endpoints.
List of Packages we are going to use:
github.com/gin-gonic/gin
github.com/sirupsen/logrus
go.temporal.io/sdk
google.golang.org/api
Goroutine
Goroutine is a lightweight thread in Golang. All programs executed by Golang run on the Goroutine. That is, the main function is also executed on the Goroutine.
In other words, every program in Golang must have a least one Goroutine.
In Golang, you can use the Goroutine to execute the function with the go keyword like the below.
Temporal
A Temporal Application is a set of Temporal Workflow Executions. Each Temporal Workflow Execution has exclusive access to its local state, executes concurrently to all other Workflow Executions, and communicates with other Workflow Executions and the environment via message passing.
A Temporal Application can consist of millions to billions of Workflow Executions. Workflow Executions are lightweight components. A Workflow Execution consumes few compute resources; in fact, if a Workflow Execution is suspended, such as when it is in a waiting state, the Workflow Execution consumes no compute resources at all.
main.go
package main
import (
"github.com/gin-gonic/gin"
"personalproject/temporal/worker"
)
func main() {
r := gin.Default()
channel1 := make(chan interface{})
defer func() {
channel1 <- struct{}{}
}()
go iamWorkFlowInitialize(channel1)
r.POST("/iambinding", worker.IamWorkFlow)
r.Run()
}
func iamWorkFlowInitialize(channel <-chan interface{}) {
err := worker.IamWorker.Run(channel)
if err != nil {
panic(err)
}
}
we will be running the temporal worker as a thread to intialize the worker and starting our Gin server in parallel.
Temporal Worker
In day-to-day conversations, the term Worker is used to denote either a Worker Program, a Worker Process, or a Worker Entity. Temporal documentation aims to be explicit and differentiate between them.
worker/worker.go
package worker
import (
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"os"
)
const IAMTASKQUEUE = "IAM_TASK_QUEUE"
var IamWorker worker.Worker = newWorker()
func newWorker() worker.Worker {
opts := client.Options{
HostPort: os.Getenv("TEMPORAL_HOSTPORT"),
}
c, err := client.NewClient(opts)
if err != nil {
panic(err)
}
w := worker.New(c, IAMTASKQUEUE, worker.Options{})
w.RegisterWorkflow(IamBindingGoogle)
w.RegisterActivity(AddIAMBinding)
return w
}
The IamBindingGoogle workFlow and AddIAMBinding Activity is registered in the Worker.
Workflow Definition refers to the source for the instance of a Workflow Execution, while a Workflow Function refers to the source for the instance of a Workflow Function Execution.
The purpose of an Activity is to execute a single, well-defined action (either short or long running), such as calling another service, transcoding a media file, or sending an email.
worker/iam_model.go
package worker
type IamDetails struct {
ProjectID string `json:"project_id"`
User string `json:"user"`
Role string `json:"role"`
}
This defines the schema of the Iam Inputs.
worker/base.go
package worker
import (
"bytes"
"encoding/json"
"fmt"
"github.com/gin-gonic/gin"
"io"
)
func LoadData(c *gin.Context, model interface{}) error {
var body bytes.Buffer
if _, err := io.Copy(&body, c.Request.Body); err != nil {
customErr := fmt.Errorf("response parsing failed %w", err)
return customErr
}
_ = json.Unmarshal(body.Bytes(), &model)
return nil
}
LoadData function is used to Unmarshal the data that is recieved in the Api request.
worker/workflowsvc.go
package worker
import (
"context"
"go.temporal.io/sdk/client"
"os"
)
var (
IamSvc IamServiceI = &iamServiceStruct{}
)
type IamServiceI interface {
IamBindingService(details IamDetails) error
}
type iamServiceStruct struct {
}
type iamServiceModel struct {
client client.Client
workflowID string
}
func (*iamServiceStruct) IamBindingService(details IamDetails) error {
cr := new(iamServiceModel)
opts := client.Options{
HostPort: os.Getenv("TEMPORAL_HOSTPORT"),
}
c, err := client.NewClient(opts)
if err != nil {
panic(err)
}
cr.client = c
workflowOptions := client.StartWorkflowOptions{
TaskQueue: IAMTASKQUEUE,
}
_, err = cr.client.ExecuteWorkflow(context.Background(), workflowOptions, IamBindingGoogle, details)
if err != nil {
return err
}
return nil
}
here is the service layer of the WorkFlow where there is an interface which implements the methods which is defined on the interface.
worker/workflow.go
package worker
import (
"github.com/gin-gonic/gin"
"github.com/sirupsen/logrus"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
"net/http"
"time"
)
func IamWorkFlow(c *gin.Context) {
var details IamDetails
err := LoadData(c, &details)
if err != nil {
logrus.Error(err)
c.JSON(http.StatusBadRequest, err)
return
}
err = IamSvc.IamBindingService(details)
if err != nil {
logrus.Error(err)
c.JSON(http.StatusBadRequest, err)
return
}
c.JSON(http.StatusOK, err)
}
func IamBindingGoogle(ctx workflow.Context, details IamDetails) (string, error) {
iamCtx := workflow.WithActivityOptions(
ctx,
workflow.ActivityOptions{
StartToCloseTimeout: 1 * time.Hour,
ScheduleToCloseTimeout: 1 * time.Hour,
RetryPolicy: &temporal.RetryPolicy{
MaximumAttempts: 3,
},
TaskQueue: IAMTASKQUEUE,
})
err := workflow.ExecuteActivity(iamCtx, AddIAMBinding, details).Get(ctx, nil)
return "", err
}
A Workflow Execution effectively executes once to completion, while a Workflow Function Execution occurs many times during the life of a Workflow Execution.
The IamBindingGoogle WorkFlow has been using the context of workflow and the iamDetails which contains information of google_project_id, user_name and the role that should be given in gcp. Those details will be send to an activity function which executes IAM Binding.
The ExecuteActivity function should have the Activity options such as StartToCloseTimeout, ScheduleToCloseTimeout, Retry policy and TaskQueue. Each Activity function can return the output that is defined the Activity.
worker/activity.go
package worker
import (
"context"
"flag"
"fmt"
"github.com/sirupsen/logrus"
"google.golang.org/api/cloudresourcemanager/v1"
"google.golang.org/api/option"
"os"
"strings"
"time"
)
func AddIAMBinding(details IamDetails) error {
projectID := details.ProjectID
member := fmt.Sprintf("user:%s", details.User)
flag.Parse()
var role string = details.Role
ctx1 := context.TODO()
crmService, err := cloudresourcemanager.NewService(ctx1, option.WithCredentialsFile(os.Getenv("GOOGLE_APPLICATION_CREDENTIALS")))
if err != nil {
logrus.Errorf("cloudresourcemanager.NewService: %v", err)
return err
}
addBinding(crmService, projectID, member, role)
policy := getPolicy(crmService, projectID)
var binding *cloudresourcemanager.Binding
for _, b := range policy.Bindings {
if b.Role == role {
binding = b
break
}
}
fmt.Println("Role: ", binding.Role)
fmt.Print("Members: ", strings.Join(binding.Members, ", "))
removeMember(crmService, projectID, member, role)
return nil
}
func addBinding(crmService *cloudresourcemanager.Service, projectID, member, role string) {
policy := getPolicy(crmService, projectID)
var binding *cloudresourcemanager.Binding
for _, b := range policy.Bindings {
if b.Role == role {
binding = b
break
}
}
if binding != nil {
binding.Members = append(binding.Members, member)
} else {
binding = &cloudresourcemanager.Binding{
Role: role,
Members: []string{member},
}
policy.Bindings = append(policy.Bindings, binding)
}
setPolicy(crmService, projectID, policy)
}
func removeMember(crmService *cloudresourcemanager.Service, projectID, member, role string) {
policy := getPolicy(crmService, projectID)
var binding *cloudresourcemanager.Binding
var bindingIndex int
for i, b := range policy.Bindings {
if b.Role == role {
binding = b
bindingIndex = i
break
}
}
if len(binding.Members) == 1 {
last := len(policy.Bindings) - 1
policy.Bindings[bindingIndex] = policy.Bindings[last]
policy.Bindings = policy.Bindings[:last]
} else {
var memberIndex int
for i, mm := range binding.Members {
if mm == member {
memberIndex = i
}
}
last := len(policy.Bindings[bindingIndex].Members) - 1
binding.Members[memberIndex] = binding.Members[last]
binding.Members = binding.Members[:last]
}
setPolicy(crmService, projectID, policy)
}
func getPolicy(crmService *cloudresourcemanager.Service, projectID string) *cloudresourcemanager.Policy {
ctx := context.Background()
ctx, cancel := context.WithTimeout(ctx, time.Second*10)
defer cancel()
request := new(cloudresourcemanager.GetIamPolicyRequest)
policy, err := crmService.Projects.GetIamPolicy(projectID, request).Do()
if err != nil {
logrus.Errorf("Projects.GetIamPolicy: %v", err)
}
return policy
}
func setPolicy(crmService *cloudresourcemanager.Service, projectID string, policy *cloudresourcemanager.Policy) {
ctx := context.Background()
ctx, cancel := context.WithTimeout(ctx, time.Second*10)
defer cancel()
request := new(cloudresourcemanager.SetIamPolicyRequest)
request.Policy = policy
policy, err := crmService.Projects.SetIamPolicy(projectID, request).Do()
if err != nil {
logrus.Errorf("Projects.SetIamPolicy: %v", err)
}
}
Google Cloud Go SDK is used here for actual iamBinding.
Finally we need temporal setup using docker,
.local/quickstart.yml
version: '3.2'
services:
elasticsearch:
container_name: temporal-elasticsearch
environment:
- cluster.routing.allocation.disk.threshold_enabled=true
- cluster.routing.allocation.disk.watermark.low=512mb
- cluster.routing.allocation.disk.watermark.high=256mb
- cluster.routing.allocation.disk.watermark.flood_stage=128mb
- discovery.type=single-node
- ES_JAVA_OPTS=-Xms100m -Xmx100m
volumes:
- esdata:/usr/share/elasticsearch/data:rw
image: elasticsearch:7.10.1
networks:
- temporal-network
ports:
- 9200:9200
postgresql:
container_name: temporal-postgresql
environment:
POSTGRES_PASSWORD: temporal
POSTGRES_USER: temporal
image: postgres:9.6
networks:
- temporal-network
ports:
- 5432:5432
temporal:
container_name: temporal
depends_on:
- postgresql
- elasticsearch
environment:
- DB=postgresql
- DB_PORT=5432
- POSTGRES_USER=temporal
- POSTGRES_PWD=temporal
- POSTGRES_SEEDS=postgresql
- DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development_es.yaml
- ENABLE_ES=true
- ES_SEEDS=elasticsearch
- ES_VERSION=v7
image: temporalio/auto-setup:1.13.1
networks:
- temporal-network
ports:
- 7233:7233
volumes:
- ./dynamicconfig:/etc/temporal/config/dynamicconfig
temporal-admin-tools:
container_name: temporal-admin-tools
depends_on:
- temporal
environment:
- TEMPORAL_CLI_ADDRESS=temporal:7233
image: temporalio/admin-tools:1.13.1
networks:
- temporal-network
stdin_open: true
tty: true
temporal-web:
container_name: temporal-web
depends_on:
- temporal
environment:
- TEMPORAL_GRPC_ENDPOINT=temporal:7233
- TEMPORAL_PERMIT_WRITE_API=true
image: temporalio/web:1.13.0
networks:
- temporal-network
ports:
- 8088:8088
networks:
temporal-network:
driver: bridge
intranet:
volumes:
esdata:
driver: local
Export the environment variables in terminal :
export TEMPORAL_HOSTPORT=localhost:7233
export GOOGLE_APPLICATION_CREDENTIALS={{path of your SPN File}}
Run the docker-compose file to start the temporal :
docker-compose -f .local/quickstart.yml up --build --force-recreate -d
Perfect!! We are all set now. Let’s run this project:
go run main.go
And I can see an Engine instance has been created and the APIs are running and the temporal is started as a thread:
Running Gin server…
And Even the Temporal UI is on http://localhost:8088
Let’s hit our POST API:
So, It’s a SUCCESS!
The Workflow is completed and IamBinding is Done is GCP also.
If you find any kind of difficulty following the above steps, please check this repo and run:
git clone https://github.com/venkateshsuresh/temporal-iamBinding-GCP-workflow.git
I hope this article helped you. Thanks for reading and stay tuned!