conductor-sdk / conductor-go Goto Github PK

Test coverage goal:

• Authentication is supported for all the APIs
• Update task by reference name (from Orkes playground)
• Optional metrics usage
  • Avoid metrics server at opt out case
• Recent changes for Python SDK

Resources

• https://swagger.io/tools/swagger-codegen/

Similar to code found at Conductor Java SDK

Resources

• https://netflix.github.io/conductor/metrics/client/

When using taskRunner.WaitWorkers() I was curious if there is a way to gracefully shut the task runner down.
The documentation says:

WaitWorkers uses an internal waitgroup to block the calling thread until all workers started by this TaskRunner have been stopped.

Which implies, that after stopping all the workers the function WaitWorkers should stop blocking the calling thread.
After looking at the source code (sdk/worker/task_runner.go) I found out, that "stopping" a worker means setting it's batch size to zero. This correctly logs to the console, that the worker has been stopped, but the function WaitWorkers was still blocking the calling thread.

From inspecting the source code this behavior makes sense:

1. The WaitWorkers function simply waits for the internal wait group to be empty/done.

   func (c *TaskRunner) WaitWorkers() {
       c.workerWaitGroup.Wait()
   }

2. When a worker is started c.workerWaitGroup.Add(1) and go c.work4ever() are executed.
3. The function work4ever handles the Done signal to the c.workerWaitGroup like this:

   func (c *TaskRunner) work4ever(taskName string, executeFunction model.ExecuteTaskFunction, domain string) {
       defer c.workerWaitGroup.Done()
       defer concurrency.HandlePanicError("poll_and_execute")
       for c.isWorkerRegistered(taskName) {
           c.workOnce(taskName, executeFunction, domain)
       }
   }

   This means that c.workerWaitGroup.Done() is only called when the for loop is done. The for loop only stops when c.isWorkerRegistered(taskName) returns false. So let's have a look at this function.
4. Looking at the implementation of isWorkerRegistered it's clear, that the batch size of 0 does not affect the return value of isWorkerRegistered because it only checks if the task name is present as a key in the map.

   func (c *TaskRunner) isWorkerRegistered(taskName string) bool {
       c.batchSizeByTaskNameMutex.RLock()
       defer c.batchSizeByTaskNameMutex.RUnlock()
       _, ok := c.batchSizeByTaskName[taskName]
       return ok
   }

If this is the expected behavior, the documentation should clearly differentiate between a registered worker and a stopped worker. Also the difference between these two "states" and a paused worker is not clear to me.

In my case I want to dynamically modify the task runner via an REST API. This means adding and "removing" workers at runtime. As there is no clear way on how to "remove" a worker from a task runner I'm unsure if pausing is sufficient, if I should set the batch size to 0 or if I need to create a PR and add a RemoveWorker(...) functionality which actually deletes task name key from the batchSizeByTaskName map.

Hopefully someone can help me out on this issue and shine some light on the differences between the different "states" of a worker.

Edit: I have an additional question. Why is there a possible goroutine leak when calling Pause(taskName)?

Related issue: https://github.com/orkes-io/orkes-conductor/issues/111

Evaluate behavior on invalid credentials:

• Should workers keep polling with invalid token?

Change parallel go routines amount to parallels routine count

Issue:

Even though batch size is set as 1. Noticed that 2 tasks were dequeued by the worker.

RCA:

There is a race between getAvailableWorkerAmount() and executeAndUpdateTask() in workOnce() code. Here are the sequence of events:

STEPS to reproduced
BATCH_SIZE is set to 1 in my case.

1. More than 1 tasks are present in the work queue.
2. Worker is started and a task is dequeued and processing started but not finished yet for first task.
3. But Worker dequeuing another task without checking as first task still running

FIX
Conductor GO SDK need to fix and below PR raised for same . Please take a look and merge it and new release .or let me know whats the process to get it fixed on condutor-go as its blocking our release.
#118

conductor version
conductor-go v1.3.2

Note:
it seems like the Java SDK client has moved away from this design and using TaskRunnerConfig to poll for tasks. But go client isn't quite there yet. But as this is blocking our release we fixed this issue on conductor-go client in below PR
#118

As a user, I would like to have the easiest possible onboard.

Some users might face missing imports. Consider reviewing current guide and add necessary steps as needed. Probably worth mentioning about go mod tidy - it might solve most problems related to missing package or failed import.

Accept-Encoding: gzip

Replace sleep(1) approach to use channels instead

Use Orkes Playground Swagger docs as reference: https://play-app.orkes.io/api-docs

Steps:

• Generate code using latest version of Swagger Code Generator
• Replace current http client with new code and fix diffs

Hello! Just wondering if there is a function available to parse a JSON workflow into a typed ConductorWorkflow object? Thank you!

Hi ,

I am running conductor as docker-compose in local, and wrote a simple program in go for worker - https://github.com/sandeep540/netflix_conductor

It's running, but not completing task and there are no logs, can you please help me in debuggin where am I going wrong,

P.S: I got it working with java, but we have requirement with go

Hi, just wonder is there any way to create a task with retryCount=0? CauseretryCount is an "omitempty" attribute.

Also try some other value like retryCount=-1 etc. however there's validation on server side

Maybe using *int32 for retryCount is a better way in TaskDef?

Hi

I am using an API - workflowExecutor.UpdateTask() to pass task status and output, but output is not going to conductor

It's failing at