Fix initial dask job state

go/tasks/plugins/k8s/dask/dask.go

@@ -320,7 +320,10 @@ func (p daskResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s
 		OccurredAt: &occurredAt,
 	}
 
-	isQueued := status == daskAPI.DaskJobCreated ||
+	// There is a short period between the `DaskJob` resource being created and `Status.JobStatus` being set by the `dask-operator`.
+	// In that period, the `JobStatus` will be an empty string. We're treating this as Initializing/Queuing.
+	isQueued := status == "" ||
+		status == daskAPI.DaskJobCreated ||
 		status == daskAPI.DaskJobClusterCreated
 
 	if !isQueued {
@@ -337,6 +340,8 @@ func (p daskResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s
 	}
 
 	switch status {
+	case "":
+		return pluginsCore.PhaseInfoInitializing(occurredAt, pluginsCore.DefaultPhaseVersion, "unknown", &info), nil
 	case daskAPI.DaskJobCreated:
 		return pluginsCore.PhaseInfoInitializing(occurredAt, pluginsCore.DefaultPhaseVersion, "job created", &info), nil
 	case daskAPI.DaskJobClusterCreated:

go/tasks/plugins/k8s/dask/dask_test.go

@@ -491,7 +491,14 @@ func TestGetTaskPhaseDask(t *testing.T) {
 	daskResourceHandler := daskResourceHandler{}
 	ctx := context.TODO()
 
-	taskPhase, err := daskResourceHandler.GetTaskPhase(ctx, nil, dummyDaskJob(daskAPI.DaskJobCreated))
+	taskPhase, err := daskResourceHandler.GetTaskPhase(ctx, nil, dummyDaskJob(""))
+	assert.NoError(t, err)
+	assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseInitializing)
+	assert.NotNil(t, taskPhase.Info())
+	assert.Nil(t, taskPhase.Info().Logs)
+	assert.Nil(t, err)
+
+	taskPhase, err = daskResourceHandler.GetTaskPhase(ctx, nil, dummyDaskJob(daskAPI.DaskJobCreated))
 	assert.NoError(t, err)
 	assert.Equal(t, taskPhase.Phase(), pluginsCore.PhaseInitializing)
 	assert.NotNil(t, taskPhase.Info())