-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
(feat): Cron Trigger Service Capability
- Loading branch information
1 parent
ad84133
commit 47b94dc
Showing
6 changed files
with
612 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,287 @@ | ||
| package triggers | ||
|
|
||
| import ( | ||
| "context" | ||
| "crypto/sha256" | ||
| "encoding/hex" | ||
| "errors" | ||
| "fmt" | ||
| "sync" | ||
| "time" | ||
|
|
||
| // actually uses cronv3 internally for its | ||
| // parser to turn cron strings into cron schedules | ||
| "github.com/go-co-op/gocron/v2" | ||
| "github.com/robfig/cron/v3" | ||
|
|
||
| "github.com/smartcontractkit/chainlink-common/pkg/capabilities" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/logger" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/services" | ||
| "github.com/smartcontractkit/chainlink-common/pkg/values" | ||
| ) | ||
|
|
||
| const cronTriggerID = "[email protected]" | ||
|
|
||
| var cronTriggerInfo = capabilities.MustNewCapabilityInfo( | ||
| cronTriggerID, | ||
| capabilities.CapabilityTypeTrigger, | ||
| "A trigger to schedule workflow execution to run periodically at fixed times, dates, or intervals.", | ||
| ) | ||
|
|
||
| type cronTriggerConfig struct { | ||
| // An identifier to register this trigger under, must be unique across all triggers within a workflow. | ||
| TriggerID string `json:"triggerId"` | ||
| // The crontab to evaluate, with second granularity. | ||
| // The seconds field is required. The time zone will always be automatically set to UTC. | ||
| Schedule string `json:"schedule"` | ||
| } | ||
|
|
||
| type cronTriggerInput struct{} | ||
|
|
||
| type CronTriggerPayload struct { | ||
| ScheduledExecutionTime string | ||
| } | ||
|
|
||
| type cronJob struct { | ||
| ch chan<- capabilities.CapabilityResponse | ||
| job gocron.Job | ||
| } | ||
|
|
||
| type CronTriggerService struct { | ||
| capabilities.CapabilityInfo | ||
| capabilities.Validator[cronTriggerConfig, cronTriggerInput, capabilities.CapabilityResponse] | ||
| jobs map[string]cronJob | ||
| lggr logger.Logger | ||
| mu sync.Mutex | ||
| scheduler gocron.Scheduler | ||
| stopCh services.StopChan | ||
| wg sync.WaitGroup | ||
| } | ||
|
|
||
| var _ capabilities.TriggerCapability = (*CronTriggerService)(nil) | ||
| var _ services.Service = &CronTriggerService{} | ||
|
|
||
| func NewCronTriggerService(lggr logger.Logger) *CronTriggerService { | ||
| s, err := gocron.NewScheduler() | ||
| if err != nil { | ||
| return nil | ||
| } | ||
| return &CronTriggerService{ | ||
| CapabilityInfo: cronTriggerInfo, | ||
| Validator: capabilities.NewValidator[cronTriggerConfig, cronTriggerInput, capabilities.CapabilityResponse](capabilities.ValidatorArgs{Info: cronTriggerInfo}), | ||
| jobs: map[string]cronJob{}, | ||
| lggr: logger.Named(lggr, "CronTrigger"), | ||
| scheduler: s, | ||
| stopCh: make(services.StopChan), | ||
| } | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) RegisterTrigger(ctx context.Context, req capabilities.CapabilityRequest) (<-chan capabilities.CapabilityResponse, error) { | ||
| cts.mu.Lock() | ||
| defer cts.mu.Unlock() | ||
|
|
||
| config, err := cts.ValidateConfig(req.Config) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| triggerID := getTriggerID(config.TriggerID, req.Metadata.WorkflowID) | ||
| if _, ok := cts.jobs[triggerID]; ok { | ||
| return nil, fmt.Errorf("triggerId %s already registered", triggerID) | ||
| } | ||
|
|
||
| job := gocron.CronJob(config.Schedule, true) | ||
|
|
||
| cronParser := cron.NewParser( | ||
| cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow, | ||
| ) | ||
| schedule, err := cronParser.Parse(config.Schedule) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| t := gocron.NewTask( | ||
| cts.makeProcess(triggerID, schedule), | ||
| ) | ||
|
|
||
| j, err := cts.scheduler.NewJob( | ||
| job, | ||
| t, | ||
| ) | ||
|
|
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| ch := make(chan capabilities.CapabilityResponse, defaultSendChannelBufferSize) | ||
| cts.jobs[triggerID] = cronJob{ | ||
| ch: ch, | ||
| job: j, | ||
| } | ||
| cts.lggr.Debugw("RegisterTrigger", "triggerId", triggerID, "jobId", j.ID()) | ||
| return ch, nil | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) makeProcess(triggerId string, schedule cron.Schedule) func() { | ||
| return func() { cts.process(triggerId, schedule) } | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) process(triggerId string, schedule cron.Schedule) { | ||
| cts.mu.Lock() | ||
| defer cts.mu.Unlock() | ||
|
|
||
| now := time.Now() | ||
| scheduledExecutionTime := schedule.Next(now) | ||
| // if scheduledExecutionTime is before now, get the next one | ||
| // this can be due to time sync / sleep | ||
| for { | ||
| if scheduledExecutionTime.Before(now) { | ||
| scheduledExecutionTime = schedule.Next(now.Add(time.Second)) | ||
| } else { | ||
| break | ||
| } | ||
| } | ||
|
|
||
| // timestamp in ns precision, to show the discrepancy between the scheduled time and the actual time | ||
| // we need scheduled time to generate a deterministic triggerEventId | ||
| timestampNs := time.Now().UTC().UnixNano() | ||
| // format to ISO 8601 | ||
| timestampMsIso8601 := formatUnixNanoToISO8601(timestampNs) | ||
|
|
||
| scheduledExecutionTimeFormatted := scheduledExecutionTime.Format(time.RFC3339) | ||
| hash := sha256.Sum256([]byte(scheduledExecutionTimeFormatted)) | ||
| triggerEventId := hex.EncodeToString(hash[:]) | ||
|
|
||
| cts.lggr.Debugw("process", "scheduledExecTime", scheduledExecutionTime, "actualExecTime", timestampMsIso8601, "triggerEventId", triggerEventId) | ||
|
|
||
| capabilityResponse, err := wrapCronTriggerEvent(triggerEventId, timestampMsIso8601, scheduledExecutionTimeFormatted) | ||
| if err != nil { | ||
| cts.lggr.Errorw("error wrapping trigger event", "err", err) | ||
| } | ||
|
|
||
| select { | ||
| case cts.jobs[triggerId].ch <- capabilityResponse: | ||
| default: | ||
| cts.lggr.Errorw("channel full, dropping event", "eventID", triggerEventId, "triggerID", triggerID, "capabilityResponse", capabilityResponse) | ||
| } | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) UnregisterTrigger(ctx context.Context, req capabilities.CapabilityRequest) error { | ||
| cts.mu.Lock() | ||
| defer cts.mu.Unlock() | ||
|
|
||
| config, err := cts.ValidateConfig(req.Config) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| triggerID := getTriggerID(config.TriggerID, req.Metadata.WorkflowID) | ||
|
|
||
| if _, ok := cts.jobs[triggerID]; !ok { | ||
| return fmt.Errorf("triggerId %s not found", triggerID) | ||
| } | ||
|
|
||
| trigger, ok := cts.jobs[triggerID] | ||
| jobId := trigger.job.ID() | ||
|
|
||
| if ok { | ||
| cts.scheduler.RemoveJob(jobId) | ||
| close(trigger.ch) | ||
| } | ||
| delete(cts.jobs, triggerID) | ||
| cts.lggr.Debugw("UnregisterTrigger", "triggerId", triggerID, "jobId", jobId) | ||
| return nil | ||
| } | ||
|
|
||
| // Start the service. | ||
| func (cts *CronTriggerService) Start(ctx context.Context) error { | ||
| cts.wg.Add(1) | ||
|
|
||
| if cts.scheduler == nil { | ||
| return errors.New("no scheduler initialized") | ||
| } | ||
| cts.scheduler.Start() | ||
|
|
||
| cts.lggr.Info(cts.Name() + " started") | ||
|
|
||
| // block until ready to shut down | ||
| go func() { | ||
| defer cts.wg.Done() | ||
| <-cts.stopCh | ||
| }() | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Close stops the Service. | ||
| // Invariants: After this call the Service cannot be started | ||
| // again, new Service will need to be built to do so. | ||
| func (cts *CronTriggerService) Close() error { | ||
| err := cts.scheduler.Shutdown() | ||
| if err != nil { | ||
| return fmt.Errorf("scheduler shutdown encountered a problem: %s", err) | ||
| } | ||
|
|
||
| close(cts.stopCh) | ||
|
|
||
| cts.wg.Wait() | ||
|
|
||
| cts.lggr.Info(cts.Name() + " closed") | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) Ready() error { | ||
| if cts.scheduler == nil { | ||
| return errors.New("no scheduler initialized") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) HealthReport() map[string]error { | ||
| return map[string]error{cts.Name(): nil} | ||
| } | ||
|
|
||
| func (cts *CronTriggerService) Name() string { | ||
| return "CronTriggerService" | ||
| } | ||
|
|
||
| func getTriggerID(triggerID string, wid string) string { | ||
| return wid + "|" + triggerID | ||
| } | ||
|
|
||
| func wrapCronTriggerEvent(eventID string, timestamp string, scheduledExecutionTime string) (capabilities.CapabilityResponse, error) { | ||
| payload, err := values.Wrap(CronTriggerPayload{ScheduledExecutionTime: scheduledExecutionTime}) | ||
| if err != nil { | ||
| return capabilities.CapabilityResponse{}, err | ||
| } | ||
|
|
||
| metadata, err := values.Wrap(nil) | ||
| if err != nil { | ||
| return capabilities.CapabilityResponse{}, err | ||
| } | ||
|
|
||
| triggerEvent := capabilities.TriggerEvent{ | ||
| TriggerType: cronTriggerID, | ||
| ID: eventID, | ||
| Timestamp: timestamp, | ||
| Metadata: metadata, | ||
| Payload: payload, | ||
| } | ||
|
|
||
| eventVal, err := values.Wrap(triggerEvent) | ||
| if err != nil { | ||
| return capabilities.CapabilityResponse{}, err | ||
| } | ||
|
|
||
| return capabilities.CapabilityResponse{ | ||
| Value: eventVal.(*values.Map), | ||
| }, nil | ||
| } | ||
|
|
||
| func formatUnixNanoToISO8601(unixNano int64) string { | ||
| seconds := unixNano / int64(time.Second) | ||
| nanoseconds := unixNano % int64(time.Second) | ||
| t := time.Unix(seconds, nanoseconds) | ||
| return t.Format(time.RFC3339Nano) | ||
| } |
Oops, something went wrong.