How to create on-demand service clients only during a workflow trigger

Community Support
,
1

Hi all, I am new to temporal and trying to make a usecase work. i have created a public repo to easily collaborate -

I am integrating gcp provider to use the underlying go sdk to interact with all gcp services.
list of gcp services : Cloud Storage (GCS) - Package cloud.google.com/go/storage (v1.40.0)  |  Go client library  |  Google Cloud

i am creating the client directly at the worker level - which isn’t feasible because of the connection being idle , when it is not in use - and it will run throughout the lifetime of the worker, and since i will be having 100+ connections (per each gcp service) - but only one worker per gcp provider - i’d like to find a way to create the connection on on-demand basis - only when the specific gcp service workflow is triggered. so whenever , a gcp storage service is invoked, its respective workflow is triggered, a gcp storage client connection is established and using that client , perform some CRUD ops and completes the workflows - along with closing the client connection.

I’d like to get some help on how do i make these client connections on need-basis , only when the workflow is triggered for that specific gcp service. I have one worfklow per gcp service - and rest all are the activities - now i cannot create a gcp client inside the workflow (i tried - its not working) - nor can i create it for each acitvity (i will be having around 15-20 activities per workflow) - so the only option i knew was to create the gcp client at worker level and pass to the activityHandle struct receiver (pls see the codebase). now this is not at all feasible - thats why i need some guidance on how to move forward for creating gcp service client ONLY when its respective workflow is triggered .

right now - I am only using “cloudstorage” as the only gcp service - but once i manage to create client per workflow - i will be intergating all gcp services which are over 100+ .

Any kind of help is highly appreciated - also feel free to checkout the above mentioned repo and raise a PR , if you feel like it. thanks in advance.

2

i will be having around 15-20 activities per workflow

does each one needs the workflow-specific client connection?

is the connection a serializable object? if so you consider creating it once when workflow starts (maybe local activity) and then pass it to rest of activities that need it.

3

yes - each gcp service is mapped to a dedicated workflow - and logic is executed as activities.

no , the connection is not serializable object - these are all the diff ways that i’ve tried, none of them worked.

  1. created the gcp storage client at workflow level and passed it to activity - didnt work
  2. created the client in the first activity - stored it in a struct and tried to access it in another activity - didnt work
  3. created the client in the first activity - returned the client object and passed onto another activity via args - didnt work
  4. created the client in a separate func - tried calling it in both activity and workflow - didnt work
  5. lastly - i tried using localactivity to create a client and returned it to the workflow, stored the returned client object in a struct , and passed that struct to another activity - still does NOT work

The problem is that the moment i create a client and pass it to another func (workflow to activity, activity to activity, and activity to worklow) - it does not work - the client object becomes nil

can you please hlep me out here - am struggling to find a way forward. thanks

4

Can we have the code that creates your gcp client run on the activity worker
host and then route all your activities for the workflow execution to the same host? In this case you would have a single activity worker that polls on your activity task queue and in your activity options would set this task queue to route all your activity executions to same one.

5

Go SDK also has sessions api, see sample samples-go/fileprocessing at main · temporalio/samples-go · GitHub
which has advantages over the task-queue routing approach which requires single activity worker imo (not very easily scalable approach)

6

the code is already provided in the sample repo that i’ve shared. GitHub - artinhum/gcp-poc

yes - i have just one worker that is being used for the workflow and the activities - so by default all my activities + workflows will run on this single worker - so i dont think i would be needing sessions api. right ?

7

right now in the code thats shared - the gcp storage client is being created at worker level - all i want is to create this client and use it at workflow level , meaning that this client should be created ONLY AFTER the workflows starts - all the activities are executed using this single client - and client connection is closed just before the workflow completes. this is the only rrequirement i have .

8

Got it, yeah I mean you need to then either create this client in a local activity/activity or a SideEffect and then pass it to each of the activities as input. If that’s not possible maybe look at the context propagation sample samples-go/ctxpropagation at main · temporalio/samples-go · GitHub or some ways to write this client to host memory and retrieving it in activities.

9

well i already tried using local-activity/ activity - it just does not work. i dont want to use side effect as the docs says side-effect isn’t feasible to use when theres a chance of failure so i thinks its better to avoid side effect .

i already looked at the ctx propogation sample - i dont think it fits my use-case , since the context is propogated at worker level as per the sample - and as i said , i have over 100+ clients that i need - one for each gcp service - thats assosicated with a dedicated workflow - so i cannot ctx propogation.

can you please advise some other way ? or if you yourself can give it a try from the sample repo that i’ve shared ? maybe you might find something ?

10

i dont think it fits my use-case , since the context is propogated at worker level as per the sample

yes in the sample but you can propagate context from workflow to activity, in the sample you can just do
in workflow code:

ctx = workflow.WithValue(ctx, ...)

11

i forgot to mention this - but i have already tried this - its throwing me this error

2024/05/02 09:16:33 ERROR LocalActivity panic. Namespace default TaskQueue GCP WorkerID 85845@kashus-MacBook-Air.local@ WorkflowID GCP_Storage_CloudStorage RunID 36e52410-e552-4bba-82d8-156f3afa0c14 ActivityType CreateClient Attempt 1 PanicError interface conversion: interface {} is nil, not cloud_storage.gcpStorageClient PanicStack local activity for CreateClient [panic]:

I have created it like this

type gcpStorageClient struct {
  client *storage.Client
}

// i cannot use workflow.context here because its of internal.context type , and the expected here is context.context type

c, e := storage.NewClient(context.Background())
	if e!= nil {
		logger.Error("Unable to create client", "Error", e)
		return false, e
	}

	gc := &gcpStorageClient{
		client: c,
	}

	cv := workflow.WithValue(ctx, "SC", gc)
	lao := workflow.LocalActivityOptions{
		StartToCloseTimeout:    20 * time.Second,
	}
	lctx := workflow.WithLocalActivityOptions(cv, lao)
    err := workflow.ExecuteLocalActivity(lctx, CreateClient).Get(ctx, &ah)

and then inside my activity

	var scc gcpStorageClient

	scc = ctx.Value("SC").(gcpStorageClient) 
// THIS IS WHERE THE TYPE ASSERTION ERROR IS COMING IN
12

You cannot propagate storage.Client between workflow and activities as it cannot be converted to/from JSON.

I would recommend caching the client inside the activity implementation structure inside a map with WorkflowRunID as a key. Then use Session to route all the activities that use the client to the same host.

13

That is exactly where am stuck with - could you please help me get started on how to cache a client in one activity and reuse that in another activity ? am struggling to get started on that. appreciate your help here.

14

Something like this (assuming that all activities use Session to run on the same process):

type MyActivities struct {
	mu sync.Mutex
	clients map[string]Client
}

func (a *MyActivities) Activity1(ctx context.Context)  {
	client := initClient();
	runID := activity.GetInfo(ctx).WorkflowExecution.RunID
	a.mu.Lock()
	a.clients[runID] = client
	a.mu.Unlock()
	//u use client
}

func (a *MyActivities) Activity2(ctx context.Context)  {
	runID := activity.GetInfo(ctx).WorkflowExecution.RunID
	a.mu.Lock()
	client := a.clients[runID]
	a.mu.Unlock()
	// use client
}

func (a *MyActivities) Activity3(ctx context.Context) {
	runID := activity.GetInfo(ctx).WorkflowExecution.RunID
	a.mu.Lock()
	client := a.clients[runID]
	a.mu.Unlock()
	// Last activity deletes client from the map
	defer func() {
		a.mu.Lock()
		delete(a.clients, runID)
		a.mu.Unlock()
	}()
	// use client
}
1 Like
15

thanks for the help, but looks like this does not support running of multiple activities in parallel ?

16

HI Maxim - i tried using the map + mutex method that you have shared - am still facing the issue , the client object keeps on getting set to nil when tried to use it inside another activity

2024/05/02 11:15:52 ERROR Activity panic. Namespace default TaskQueue GCP WorkerID 99624@kashus-MacBook-Air.local@ WorkflowID GCP_Storage_CloudStorage RunID dcbd6183-35e6-41a5-8e81-f7f281ab1ca2 ActivityType CreateClient Attempt 1 PanicError assignment to entry in nil map PanicStack activity for 560274f2-d3bb-4e99-8034-8af8e1100f16@kashus-MacBook-Air.local [panic]:
17

thanks for the help, but looks like this does not support running of multiple activities in parallel ?

Why do you think so? It is uses Mutex to ensure safe access to the map.

HI Maxim - i tried using the map + mutex method that you have shared - am still facing the issue , the client object keeps on getting set to nil when tried to use it inside another activity

Are you using Session to route activities to the same process?

18

Hi Maxim, i was able to make it work. thank you so much for all the help, really appreciate it. :slight_smile:

1 Like
19

Hi @maxim @tihomir , I am back to the same problem again , but this time the issue is across different workflows instead of a different activities within workflow.

I need a share a client (github client) across multiple workflows (nested childworkflows) across multiple go packages. I tried creating the client in a separate git package and initiated the client at the root workflow (parent workflow) , and since its a different package i cannot create methods on top of it - so am trying to wrap it in a struct and send it via args - but its not working like you said before.

can you help me on how do i share the client across different workflows ?

note: i have one single parent workflow, which triggers childworkflow - which in turn triggers another childworkflow and so on - i have 5 level nesting - and each workflow runs in a separate go pkg because of the business logic. all these workflows need to interact with github - and i want to use a single github client throught the journey of the main parent workflow and share across all the child workflows.

20

You cannot share clients across workflows, as workflows cannot make external API calls directly. Only activities can.