Merge pull request #3171 from hashicorp/b/dynamic-configsourcer-vars-…

…before-job-assign

internal/runner: Send down dynamic ConfigSourcers on JobAssign

.changelog/3171.txt

@@ -0,0 +1,4 @@
+```release-note:bug
+core: Ensure remote runners have dynamic config sources overrides for
+evaluating defaults for job variables.
+```

internal/config/variables/variables.go

@@ -421,6 +421,7 @@ func LoadDynamicDefaults(
  ctx context.Context,
  log hclog.Logger,
  pbvars []*pb.Variable,
+ cfgSrcs []*pb.ConfigSource,
  vars map[string]*Variable,
  dynamicOpts ...appconfig.Option,
 ) ([]*pb.Variable, hcl.Diagnostics) {
@@ -439,9 +440,12 @@ func LoadDynamicDefaults(
 
  // If we have no dynamic vars, we do nothing.
  if len(dynamicVars) == 0 {
+ log.Debug("no dynamic vars")
  return nil, diags
  }
 
+ log.Debug("dynamic variables discovered", "total", len(dynamicVars))
+
  // Go through our variable values and delete any dynamic vars we have
  // values for already; we do not need to fetch those.
  for _, pbv := range pbvars {
@@ -450,6 +454,7 @@ func LoadDynamicDefaults(
 
  // If we have no dynamic vars we need values for, also do nothing.
  if len(dynamicVars) == 0 {
+ log.Debug("no dynamic vars needed values")
  return nil, diags
  }
 
@@ -489,6 +494,9 @@ func LoadDynamicDefaults(
  })
  }
 
+ // Update and send any config source overrides for dynamic vars.
+ w.UpdateSources(ctx, cfgSrcs)
+
  // Send our variables. Purposely ignore the error return value because
  // it can only ever be a context cancellation which we pick up in the
  // select later.
@@ -514,6 +522,7 @@ func LoadDynamicDefaults(
 
  case config := <-ch:
  var result []*pb.Variable
+
  for _, f := range config.Files {
  result = append(result, &pb.Variable{
  Name: f.Path,

internal/config/variables/variables_test.go

@@ -257,10 +257,12 @@ func TestVariables_LoadDynamicDefaults(t *testing.T) {
  needs := NeedsDynamicDefaults(tt.provided, vars)
  require.Equal(tt.needs, needs)
 
+ var dcv []*pb.ConfigSource
  dynVars, diags := LoadDynamicDefaults(
  context.Background(),
  hclog.L(),
  tt.provided,
+ dcv,
  vars,
  appconfig.WithPlugins(map[string]*plugin.Instance{
  "static": {

internal/runner/accept.go

@@ -427,7 +427,7 @@ RESTART_JOB_STREAM:
  // The job stream setup is done. Actually run the job, download any
  // data necessary, setup the core, etc
  log.Info("starting job execution")
- result, err := r.prepareAndExecuteJob(ctx, log, ui, &sendMutex, client, assignment.Assignment.Job)
+ result, err := r.prepareAndExecuteJob(ctx, log, ui, &sendMutex, client, assignment.Assignment)
  log.Debug("job finished", "error", err)
 
  // We won't output anything else to the UI anymore.
@@ -508,8 +508,9 @@ func (r *Runner) prepareAndExecuteJob(
  ui terminal.UI,
  sendMutex *sync.Mutex,
  client pb.Waypoint_RunnerJobStreamClient,
- job *pb.Job,
+ assignment *pb.RunnerJobStreamResponse_JobAssignment,
 ) (*pb.Job_Result, error) {
+ job := assignment.Job
  log.Trace("preparing to execute job operation", "type", hclog.Fmt("%T", job.Operation))
 
  // Some operation types don't need to download data, execute those here.
@@ -570,7 +571,7 @@ func (r *Runner) prepareAndExecuteJob(
  if err == nil {
  // Execute the job. We have to close the UI right afterwards to
  // ensure that no more output is written to the client.
- result, err = r.executeJob(ctx, log, ui, job, wd, sendMutex, client)
+ result, err = r.executeJob(ctx, log, ui, assignment, wd, sendMutex, client)
  }
  }
 

internal/runner/operation.go

@@ -35,11 +35,13 @@ func (r *Runner) executeJob(
  ctx context.Context,
  log hclog.Logger,
  ui terminal.UI,
- job *pb.Job,
+ assignment *pb.RunnerJobStreamResponse_JobAssignment,
  wd string,
  clientMutex *sync.Mutex,
  client pb.Waypoint_RunnerJobStreamClient,
 ) (*pb.Job_Result, error) {
+ job := assignment.Job
+
  // NOTE(mitchellh; krantzinator): For now, we query the project directly here
  // since we use it only in case of a missing local waypoint.hcl, and to
  // collect input variable values set on the server. I can see us moving this
@@ -191,6 +193,8 @@ func (r *Runner) executeJob(
  pbVars = append(pbVars, vcsVars...)
  pbVars = append(pbVars, job.Variables...)
 
+ log.Debug("looking to see if there are dynamic variable default values to load")
+
  // Load any dynamic default values. This happens after the above
  // because we only load dynamic default values for variables we do
  // not have values for.
@@ -207,23 +211,32 @@ func (r *Runner) executeJob(
  )
  }
 
+ log.Debug("loading default values for dynamic variables")
+
  dynamicVars, diags := variables.LoadDynamicDefaults(
  ctx,
  log,
  pbVars,
+ assignment.ConfigSources,
  cfg.InputVariables,
  appconfig.WithLogger(log),
  appconfig.WithPlugins(r.configPlugins),
+ appconfig.WithDynamicEnabled(true), // true because we've already determined variables need dynamic defaults
  )
  if diags.HasErrors() {
+ log.Warn("failed to load dynamic defaults for variables", "diags", diags)
  return nil, diags
  }
+
  if len(dynamicVars) > 0 {
+ log.Debug("dynamic variables discovered, adding to project variables")
  // If we have dynamic variable values, we _prepend_ them so that
  // they have the lowest precedence. In reality, this shouldn't
  // matter because we only grab dynamic values for vars that have
  // no value set, but we might as well be careful.
  pbVars = append(dynamicVars, pbVars...)
+ } else {
+ log.Debug("no dynamic variables found")
  }
  }