controller inserts into requests db

block · Apr 12, 2024 · d42180a · d42180a
1 parent 74e6afe
commit d42180a
Show file tree

Hide file tree

Showing 6 changed files with 18 additions and 37 deletions.
diff --git a/backend/controller/cronjobs/cronjobs.go b/backend/controller/cronjobs/cronjobs.go
@@ -15,7 +15,6 @@ import (
 	"github.com/TBD54566975/ftl/internal/cron"
 	"github.com/TBD54566975/ftl/internal/log"
 	"github.com/TBD54566975/ftl/internal/model"
-	"github.com/TBD54566975/ftl/internal/rpc/headers"
 	"github.com/TBD54566975/ftl/internal/slices"
 	"github.com/alecthomas/atomic"
 	"github.com/alecthomas/types/optional"
@@ -58,7 +57,6 @@ type hashRingState struct {
 
 type DAL interface {
 	GetCronJobs(ctx context.Context) ([]dal.CronJob, error)
-	CreateCronRequest(ctx context.Context, job dal.CronJob, addr string) (model.RequestKey, error)
 	StartCronJobs(ctx context.Context, jobs []dal.CronJob) (attemptedJobs []dal.AttemptedCronJob, err error)
 	EndCronJob(ctx context.Context, job dal.CronJob, next time.Time) (dal.CronJob, error)
 	GetStaleCronJobs(ctx context.Context, duration time.Duration) ([]dal.CronJob, error)
@@ -70,7 +68,7 @@ type Scheduler interface {
 }
 
 type CallExecuter interface {
-	Call(ctx context.Context, req *connect.Request[ftlv1.CallRequest]) (*connect.Response[ftlv1.CallResponse], error)
+	CallWithRequest(ctx context.Context, req *connect.Request[ftlv1.CallRequest], requestKey optional.Option[model.RequestKey], requestSource string) (*connect.Response[ftlv1.CallResponse], error)
 }
 
 type Service struct {
@@ -193,18 +191,13 @@ func (s *Service) executeJob(ctx context.Context, job dal.CronJob) {
 		Body: requestJSON,
 	})
 
-	requestKey, err := s.dal.CreateCronRequest(ctx, job, s.originURL.Host)
+	requestKey := model.NewRequestKey(model.OriginCron, fmt.Sprintf("%s-%s", job.Module, job.Verb))
+
+	callCtx, cancel := context.WithTimeout(ctx, s.config.Timeout)
+	defer cancel()
+	_, err = s.executor.CallWithRequest(callCtx, req, optional.Some(requestKey), s.originURL.Host)
 	if err != nil {
 		logger.Errorf(err, "failed to execute cron job %v:%v", job.DeploymentKey, job.Verb)
-	} else {
-		headers.SetRequestName(req.Header(), requestKey)
-
-		callCtx, cancel := context.WithTimeout(ctx, s.config.Timeout)
-		defer cancel()
-		_, err = s.executor.Call(callCtx, req)
-		if err != nil {
-			logger.Errorf(err, "failed to execute cron job %v:%v", job.DeploymentKey, job.Verb)
-		}
 	}
 
 	schedule, err := cron.Parse(job.Schedule)

diff --git a/backend/controller/cronjobs/cronjobs_test.go b/backend/controller/cronjobs/cronjobs_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/TBD54566975/ftl/internal/model"
 	"github.com/TBD54566975/ftl/internal/slices"
 	"github.com/alecthomas/assert/v2"
+	"github.com/alecthomas/types/optional"
 	"github.com/benbjohnson/clock"
 	"github.com/jpillora/backoff"
 )
@@ -62,10 +63,6 @@ func (d *mockDAL) indexForJob(job dal.CronJob) (int, error) {
 	return -1, fmt.Errorf("job not found")
 }
 
-func (d *mockDAL) CreateCronRequest(ctx context.Context, job dal.CronJob, addr string) (model.RequestKey, error) {
-	return model.NewRequestKey(model.OriginCron, fmt.Sprintf("%s-%s", job.Module, job.Verb)), nil
-}
-
 func (d *mockDAL) StartCronJobs(ctx context.Context, jobs []dal.CronJob) (attemptedJobs []dal.AttemptedCronJob, err error) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
@@ -147,7 +144,7 @@ type mockExecutor struct {
 	clock         *clock.Mock
 }
 
-func (e *mockExecutor) Call(ctx context.Context, req *connect.Request[ftlv1.CallRequest]) (*connect.Response[ftlv1.CallResponse], error) {
+func (e *mockExecutor) CallWithRequest(ctx context.Context, req *connect.Request[ftlv1.CallRequest], requestKey optional.Option[model.RequestKey], requestSource string) (*connect.Response[ftlv1.CallResponse], error) {
 	verbRef := schema.RefFromProto(req.Msg.Verb)
 
 	e.lock.Lock()

diff --git a/backend/controller/dal/dal.go b/backend/controller/dal/dal.go
@@ -1058,10 +1058,6 @@ func (d *DAL) CreateRequest(ctx context.Context, key model.RequestKey, addr stri
 	return d.db.CreateRequest(ctx, sql.OriginIngress, key, addr)
 }
 
-func (d *DAL) CreateCronRequest(ctx context.Context, job CronJob, addr string) (model.RequestKey, error) {
-	key := model.NewRequestKey(model.OriginCron, fmt.Sprintf("%s-%s", job.Module, job.Verb))
-	err := d.db.CreateIngressRequest(ctx, sql.OriginCron, key, addr)
-	return key, err
 func (d *DAL) GetIngressRoutes(ctx context.Context, method string) ([]IngressRoute, error) {
 	routes, err := d.db.GetIngressRoutes(ctx, method)
 	if err != nil {

diff --git a/backend/controller/sql/querier.go b/backend/controller/sql/querier.go
diff --git a/backend/controller/sql/queries.sql b/backend/controller/sql/queries.sql
@@ -407,10 +407,6 @@ VALUES ((SELECT id FROM deployments WHERE deployments.key = sqlc.arg('deployment
 INSERT INTO requests (origin, "key", source_addr)
 VALUES ($1, $2, $3);
 
--- name: CreateCronRequest :exec
-INSERT INTO requests (origin, "key", source_addr)
-VALUES ($1, $2, $3);
-
 -- name: UpsertController :one
 INSERT INTO controller (key, endpoint)
 VALUES ($1, $2)

diff --git a/backend/controller/sql/queries.sql.go b/backend/controller/sql/queries.sql.go