RSDK-10965 Adding the jobmanager to RDK #5104

bkaravan · 2025-07-01T20:00:37Z

A draft PR to get feedback on the Job Manager logic.

From the ticket, the job manager:

Should have a concept of a “job”
Should add or remove jobs according to mutations of the jobs field in the config
Should log when jobs trigger and whether they succeed or fail
Should gracefully shutdown along with machine
Should have tests

Here is how this PR answers these goals:

We rely on the "job" definition from gocron. While we considered creating an internal "Job" struct, it seemed to be a layer of abstraction that did not provide any useful features. The JobConfig already has most of the required data to start the job, so a separate struct seemed unnecessary.
When the config file gets updated, the diff gets populates Added, Removed, and Modified fields with their respective JobConfig. After that, jobmanager.UpdateJobs() is called within reconfigure, which is responsible for correctly stopping/starting jobs.
For logging patterns, explore jobmanager.go
A Shutdown() method is introduced and is called at the end of localrobot.Close()
Tests are still to be created. There are tests for config-related changes, but not for the jobmanager logic. Will be done after feedback for this part settles!

We have also decided to not use the cronexpr library to parse the cron expression during Validation. cronexpr could allow a 7th year field, while gocron does not support that. Currently, Validation only checks the required fields of the JobConfig, and any "bad schedule" messages are displayed when the jobmanager is trying to actually start the job (in case of an error, no job will be created)

config/config.go

robot/impl/local_robot.go

benjirewis · 2025-07-01T20:55:42Z

robot/jobmanager/jobmanager.go

+) (*Jobmanager, error) {
+	jobLogger := logger.Sublogger("job_manager")
+	// we do not want deduplication on jobs
+	jobLogger.NeverDeduplicate()


Per offline discussion, I recommended that we never deduplicate for jobs so that jobs that run every 1s, e.g., are visible in their logging and not squashed unnecessarily.

robot/jobmanager/jobmanager.go

benjirewis · 2025-07-01T21:08:44Z

robot/jobmanager/jobmanager.go

+			toPrint := buffer.String()
+			toPrint = strings.ReplaceAll(toPrint, "\n", "")
+			toPrint = strings.ReplaceAll(toPrint, "\"", "")
+			jm.logger.CInfow(jm.ctx, "Job succeeded", "name", jc.Name, "result", toPrint)


Could you give an example of what this log looks like when it's output?

Totally. Here is how it looks after replacements:
2025-07-01T21:20:07.714Z INFO rdk.job_manager jobmanager/jobmanager.go:172 Job succeeded {"name":"my motor job","result":"{ isOn: false, powerPct: 0}"}
Here is how it looks without these strings replacements:
2025-07-01T19:58:50.847Z INFO rdk.job_manager jobmanager/jobmanager.go:171 Job succeeded {"name":"my motor job","result":"{\n \"isOn\": false,\n \"powerPct\": 0\n}\n"}

Seems good to me! Could we change result to be response so it's clear it's a gRPC response?

Totally, will do tomorrow

If buffer is guaranteed to be JSON, could you unmarshal it into a struct and pass that into the "w" loggers, or maybe use a custom String()? I feel ReplaceAll to transform json is usually a last resort in many languages, because it's slow (generally speaking -- maybe not applicable here) and if the response can contain some strings that would be more useful if they're kept as-is.

I had an attempt where I called json.Unmarshall() directly, but could not quite get it to work. I am happy to give it another try though, currently having some laptop issues but I will give it another go tomorrow.

On that note, do you know if go will support deserializing it into a generic struct? My understanding was that you need to know the struct fields you are getting beforehand, and pass that as an argument to Unmarshall()

Not sure, but I just glanced through at the InvokeRPC src and you might even be able to get the response back as the native proto.Message via handler.OnReceiveResponse. That might be pretty useful to have, even just for additional logging flexibility.

Will explore!

web/server/entrypoint.go

benjirewis · 2025-07-01T21:12:25Z

Awesome job! Some initial questions + nits.

bkaravan · 2025-07-01T21:37:18Z

robot/impl/local_robot.go

+
+	jobManager, err := jobmanager.New(ctx, logger, getResource, r.webSvc.ModuleAddresses())
+	if err != nil {
+		logger.Warn("Unable to start the job scheduler")


Just noticed this logger line as well. What should be the correct thing to do here - do we log an error or a warning and return nil, err; or is this error going to get logged anyway when we return it?

Good point; you probably don't need the warn log line. I believe viam-server would fail to start up with error serving web: [error from failure to create job manager], but you could double-check.

Makes sense, will double check this

benjirewis · 2025-07-02T14:37:43Z

robot/impl/local_robot.go

+
+	jobManager, err := jobmanager.New(ctx, logger, getResource, r.webSvc.ModuleAddresses())
+	if err != nil {
+		logger.Warn("Unable to start the job scheduler")


Good point; you probably don't need the warn log line. I believe viam-server would fail to start up with error serving web: [error from failure to create job manager], but you could double-check.

benjirewis · 2025-07-02T14:40:39Z

robot/jobmanager/jobmanager.go

+		functionToRun := func() {
+			resource, err := jm.getResource(jc.Resource)
+			if err != nil {
+				jm.logger.CWarnw(jm.ctx, "Could not get resource", "error", err)


Do you think we should give each job its own Sublogger of rdk.job_manager? That way errors like this can easily be associated with a job? We could also use WithFields to always associate a job name with logs for a certain job. I'd pick one or the other, but up to you.

Do you think we would create a Sublogger inside the functionToRun()? Not sure how expensive a call to Sublogger is, especially on quick jobs, it might be setting up/tearing it down quite often. Could that be a reason to create a Job struct, that would hold a logger and the function for each job? Potentially, WithFields would be a better fit to not add this other layer of abstraction

I would say let's avoid creating a Sublogger in every functionToRun() call. Whether that requires a Job struct for another layer of abstraction I'm not sure. Feel free to play around with it, but I think a Job being a function and a logging.Logger is a fair, new type to add if need be.

benjirewis · 2025-07-02T14:41:43Z

robot/jobmanager/jobmanager.go

+				jm.logger.CWarnw(jm.ctx, "Could not get resource", "error", err)
+				return
+			}
+			jm.logger.CInfow(jm.ctx, "Job triggered", "name", jc.Name)


If you use a Sublogger or a WithFields logger as I suggest above, you probably don't need the "name" field here or anywhere else.

benjirewis · 2025-07-02T14:42:44Z

robot/jobmanager/jobmanager.go

+			toPrint := buffer.String()
+			toPrint = strings.ReplaceAll(toPrint, "\n", "")
+			toPrint = strings.ReplaceAll(toPrint, "\"", "")
+			jm.logger.CInfow(jm.ctx, "Job succeeded", "name", jc.Name, "result", toPrint)


Seems good to me! Could we change result to be response so it's clear it's a gRPC response?

config/diff_test.go

cheukt

looks good to me so far!

cheukt · 2025-07-02T21:12:20Z

robot/jobmanager/jobmanager.go

+		return nil, err
+	}
+
+	dialAddr := "unix://" + parentAddr.UnixAddr


what if the server was running in TCP_MODE?

cheukt · 2025-07-02T21:27:00Z

robot/jobmanager/jobmanager.go

+
+// createJobFunction returns a function that the job scheduler puts on its queue.
+func (jm *Jobmanager) createJobFunction(jc config.JobConfig) func() {
+	if jc.Method == "DoCommand" {


why not just combine this with the function below?

cheukt · 2025-07-02T21:29:14Z

robot/jobmanager/jobmanager.go

+		refCtx := metadata.NewOutgoingContext(jm.ctx, nil)
+		refClient := grpcreflect.NewClientV1Alpha(refCtx, reflectpb.NewServerReflectionClient(jm.conn))
+		reflSource := grpcurl.DescriptorSourceFromServer(jm.ctx, refClient)
+		descSource := reflSource
+
+		resourceType := resource.Name().API.SubtypeName
+		services, err := descSource.ListServices()
+		if err != nil {
+			jm.logger.CWarnw(jm.ctx, "Could not get a list of available grpc services", "error", err)
+			return
+		}
+		var grpcService string
+		for _, srv := range services {
+			if strings.Contains(srv, resourceType) {
+				grpcService = srv
+				break
+			}
+		}
+		if grpcService == "" {
+			jm.logger.CWarn(jm.ctx, fmt.Sprintf("could not find a service for type: %s", resourceType))
+			return
+		}
+		grpcMethod := grpcService + "." + jc.Method


feels like a good function to abstract away into a separate function, since the logic is pretty self contained

bkaravan added 6 commits June 27, 2025 12:14

jobmanager first draft

d298251

better context, flexible services in jobTemplate

a3a6d30

remove cronexpr dependancy

5682f69

keep track of diffs for jobConfigs

a41726d

update jobmanager logic

776eec5

lint changes

38047ec

bkaravan requested review from benjirewis and a team July 1, 2025 20:00

viambot added the safe to test This pull request is marked safe to test from a trusted zone label Jul 1, 2025

benjirewis reviewed Jul 1, 2025

View reviewed changes

bkaravan commented Jul 1, 2025

View reviewed changes

benji initial comments

1ce2f6f

bkaravan requested a review from benjirewis July 2, 2025 14:21

viambot added safe to test This pull request is marked safe to test from a trusted zone and removed safe to test This pull request is marked safe to test from a trusted zone labels Jul 2, 2025

benjirewis reviewed Jul 2, 2025

View reviewed changes

aldenh-viam reviewed Jul 2, 2025

View reviewed changes

config/diff_test.go Show resolved Hide resolved

cheukt reviewed Jul 2, 2025

View reviewed changes

RSDK-10965 Adding the jobmanager to RDK #5104

Are you sure you want to change the base?

RSDK-10965 Adding the jobmanager to RDK #5104

Uh oh!

Conversation

bkaravan commented Jul 1, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

benjirewis commented Jul 1, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cheukt left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!