diff --git a/CHANGELOG.md b/CHANGELOG.md index 433c2bfa..73324bd1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ## v1.1.5 - 04 Jun 2026 +* feat(ensure-networks): Add API endpoint and CLI command to trigger it manually without restartng the service (19 minutes ago) +* feat(control-loop/ensure-networks): Add notion of control loop, ensure every 30 minutes (34 minutes ago) * fix: fix overlay neighbor replay for inactive endpoints ## v1.1.4 - 20 Mar 2026 diff --git a/client/sand/client.go b/client/sand/client.go index a2a20ee6..54220d3d 100644 --- a/client/sand/client.go +++ b/client/sand/client.go @@ -6,11 +6,12 @@ import ( "net/http" "time" + "github.com/pkg/errors" + "github.com/Scalingo/sand/api/httpresp" "github.com/Scalingo/sand/api/params" "github.com/Scalingo/sand/api/types" apptls "github.com/Scalingo/sand/utils/tls" - "github.com/pkg/errors" "crypto/tls" ) @@ -20,6 +21,7 @@ type Client interface { NetworksList(context.Context) ([]types.Network, error) NetworkCreate(context.Context, params.NetworkCreate) (types.Network, error) NetworkShow(context.Context, string) (types.Network, error) + NodeEnsureNetworkEndpoints(context.Context) error NetworkConnect(context.Context, string, params.NetworkConnect) (net.Conn, error) NetworkDelete(context.Context, string) error EndpointCreate(context.Context, params.EndpointCreate) (types.Endpoint, error) diff --git a/client/sand/node.go b/client/sand/node.go new file mode 100644 index 00000000..e2c7d1e9 --- /dev/null +++ b/client/sand/node.go @@ -0,0 +1,37 @@ +package sand + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + + "github.com/pkg/errors" + + "github.com/Scalingo/sand/api/httpresp" +) + +func (c *client) NodeEnsureNetworkEndpoints(ctx context.Context) error { + req, err := http.NewRequest("POST", fmt.Sprintf("%s/node/ensure-network-endpoints", c.url), nil) + if err != nil { + return errors.Wrapf(err, "failed to create HTTP request") + } + req = req.WithContext(ctx) + res, err := c.httpClient.Do(req) + if err != nil { + return errors.Wrap(err, "failed to execute POST /node/ensure-network-endpoints") + } + defer res.Body.Close() + + if res.StatusCode != http.StatusNoContent { + var reserr httpresp.Error + err := json.NewDecoder(res.Body).Decode(&reserr) + if err != nil { + return errors.Wrapf(err, "failed to decode JSON in errors response: %s", res.Status) + } + + return reserr + } + + return nil +} diff --git a/cmd/sand-agent-cli/main.go b/cmd/sand-agent-cli/main.go index a9740333..458ec83f 100644 --- a/cmd/sand-agent-cli/main.go +++ b/cmd/sand-agent-cli/main.go @@ -57,6 +57,10 @@ func main() { Name: "version", Action: app.Version, }, + { + Name: "node-ensure-network-endpoints", + Action: app.NodeEnsureNetworkEndpoints, + }, { Name: "network-create", Action: app.NetworkCreate, diff --git a/cmd/sand-agent-cli/node.go b/cmd/sand-agent-cli/node.go new file mode 100644 index 00000000..f51777e8 --- /dev/null +++ b/cmd/sand-agent-cli/node.go @@ -0,0 +1,23 @@ +package main + +import ( + "context" + "fmt" + + "github.com/urfave/cli" +) + +func (a *App) NodeEnsureNetworkEndpoints(c *cli.Context) error { + client, err := a.sandClient(c) + if err != nil { + return err + } + + err = client.NodeEnsureNetworkEndpoints(context.Background()) + if err != nil { + return err + } + + fmt.Println("Network endpoints ensure triggered") + return nil +} diff --git a/cmd/sand-agent/sand-agent.go b/cmd/sand-agent/sand-agent.go index c36954b3..c83a60a5 100644 --- a/cmd/sand-agent/sand-agent.go +++ b/cmd/sand-agent/sand-agent.go @@ -8,7 +8,6 @@ import ( "path/filepath" "github.com/moby/moby/pkg/reexec" - "github.com/pkg/errors" "github.com/sirupsen/logrus" "github.com/Scalingo/go-etcd-lock/v5/lock" @@ -16,10 +15,10 @@ import ( dockeripam "github.com/Scalingo/go-plugins-helpers/ipam" dockernetwork "github.com/Scalingo/go-plugins-helpers/network" dockersdk "github.com/Scalingo/go-plugins-helpers/sdk" + "github.com/Scalingo/go-utils/cronsetup" "github.com/Scalingo/go-utils/graceful" "github.com/Scalingo/go-utils/logger" "github.com/Scalingo/go-utils/logger/plugins/rollbarplugin" - "github.com/Scalingo/sand/api/params" "github.com/Scalingo/sand/api/types" "github.com/Scalingo/sand/config" "github.com/Scalingo/sand/endpoint" @@ -29,6 +28,7 @@ import ( "github.com/Scalingo/sand/network" "github.com/Scalingo/sand/network/netmanager" "github.com/Scalingo/sand/network/overlay" + "github.com/Scalingo/sand/node" "github.com/Scalingo/sand/store" apptls "github.com/Scalingo/sand/utils/tls" "github.com/Scalingo/sand/web" @@ -83,11 +83,17 @@ func main() { endpointRepository := endpoint.NewRepository(c, dataStore, managers) networkRepository := network.NewRepository(c, dataStore, managers) - err = ensureNetworks(ctx, c, networkRepository, endpointRepository) + err = node.EnsureNetworkEndpoints(ctx, c, networkRepository, endpointRepository) if err != nil { log.WithError(err).Error("fail to ensure existing networks") os.Exit(-1) } + stopEndpointEnsureCron, err := setupEndpointEnsureCron(ctx, c, networkRepository, endpointRepository) + if err != nil { + log.WithError(err).Error("fail to setup endpoint ensure cron") + os.Exit(-1) + } + defer stopEndpointEnsureCron() vctrl := web.NewVersionController(c) nctrl := web.NewNetworksController(c, networkRepository, endpointRepository, ipAllocator) @@ -96,6 +102,7 @@ func main() { sandRouter := handlers.NewRouter(log) sandRouter.Use(handlers.ErrorMiddleware) sandRouter.HandleFunc("/version", vctrl.Show).Methods("GET") + sandRouter.HandleFunc("/node/ensure-network-endpoints", nctrl.EnsureNetworkEndpoints).Methods("POST") sandRouter.HandleFunc("/networks", nctrl.List).Methods("GET") sandRouter.HandleFunc("/networks", nctrl.Create).Methods("POST") sandRouter.HandleFunc("/networks/{id}", nctrl.Show).Methods("GET") @@ -175,69 +182,17 @@ func main() { log.Info("All APIs stopped, shutting down..") } -func ensureNetworks(ctx context.Context, c *config.Config, repo network.Repository, erepo endpoint.Repository) error { - log := logger.Get(ctx) - ctx = logger.ToCtx(ctx, log) - - log.Info("Ensure networks on node") - - endpoints, err := erepo.List(ctx, map[string]string{"hostname": c.GetPeerHostname()}) - if err == store.ErrNotFound { - return nil - } - if err != nil { - return errors.Wrapf(err, "fail to list endpoints of %v", c.GetPeerHostname()) - } - - for _, endpoint := range endpoints { - log = log.WithField("endpoint_id", endpoint.ID) - ctx = logger.ToCtx(ctx, log) - if !endpoint.Active { - log.Debug("skip inactive endpoint") - continue - } - log = log.WithFields(logrus.Fields{ - "network_id": endpoint.NetworkID, - "endpoint_id": endpoint.ID, - "endpoint_netns_path": endpoint.TargetNetnsPath, - }) - log.Info("restoring endpoint") - - network, ok, err := repo.Exists(ctx, endpoint.NetworkID) - if err != nil { - return errors.Wrapf(err, "fail to get network") - } - if !ok { - log.WithError(errors.Errorf("network not found for %v", endpoint)) - continue - } - - log.Info("ensuring network") - err = repo.Ensure(ctx, network) - if err != nil { - log.WithError(err).Error("fail to ensure network") - continue - } - - endpoint, err = erepo.Activate(ctx, network, endpoint, params.EndpointActivate{ - NSHandlePath: endpoint.TargetNetnsPath, - SetAddr: true, - MoveVeth: true, - }) - if err != nil { - // if we can't activate the endpoint because the netns path doesn't exist anymore, we - // just deactivate it. Otherwise we raise an error. - if os.IsNotExist(errors.Cause(err)) { - endpoint, err = erepo.Deactivate(ctx, network, endpoint) - if err != nil { - log.WithError(err).Error("fail to deactivate endpoint") - continue - } - } else { - log.WithError(err).Error("fail to ensure endpoint") - continue - } - } - } - return nil +func setupEndpointEnsureCron(ctx context.Context, c *config.Config, repo network.Repository, erepo endpoint.Repository) (func(), error) { + return cronsetup.Setup(ctx, cronsetup.SetupOpts{ + WithoutTelemetry: true, + Jobs: []cronsetup.Job{ + { + Name: "ensure network endpoints", + Rhythm: "@every " + c.EndpointEnsureInterval.String(), + Func: func(ctx context.Context) error { + return node.EnsureNetworkEndpoints(ctx, c, repo, erepo) + }, + }, + }, + }) } diff --git a/config/config.go b/config/config.go index 80a3a4cd..0fa1197b 100644 --- a/config/config.go +++ b/config/config.go @@ -2,6 +2,7 @@ package config import ( "os" + "time" etcdutils "github.com/Scalingo/go-utils/etcd" @@ -49,6 +50,8 @@ type Config struct { EnableDockerPlugin bool `envconfig:"ENABLE_DOCKER_PLUGIN"` DockerPluginHttpPort int `default:"9998"` + EndpointEnsureInterval time.Duration `envconfig:"ENDPOINT_ENSURE_INTERVAL" default:"30m"` + MaxVNI int `envconfig:"MAX_VNI" default:"999_999"` } @@ -60,6 +63,9 @@ func Build() (*Config, error) { if err != nil { return nil, errors.Wrapf(err, "fail to build etcd config") } + if c.EndpointEnsureInterval <= 0 { + return nil, errors.New("endpoint ensure interval must be positive") + } if c.PublicHostname == "" { h, err := os.Hostname() diff --git a/go.mod b/go.mod index ec7f9df3..63db2f3c 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/Scalingo/go-etcd-lock/v5 v5.1.0 github.com/Scalingo/go-handlers v1.11.1 github.com/Scalingo/go-plugins-helpers v1.4.0 + github.com/Scalingo/go-utils/cronsetup v1.6.0 github.com/Scalingo/go-utils/etcd v1.2.2 github.com/Scalingo/go-utils/graceful v1.3.3 github.com/Scalingo/go-utils/logger v1.12.2 diff --git a/go.sum b/go.sum index 76b25e85..c9d0d8ee 100644 --- a/go.sum +++ b/go.sum @@ -9,6 +9,8 @@ github.com/Scalingo/go-handlers v1.11.1 h1:c1BnZe2EhCebIzfygoOmcl7iZlUdn5OYuG/4x github.com/Scalingo/go-handlers v1.11.1/go.mod h1:FP+IibVd+yWkSY9B+rzcIi629MMk/7kNfcEjIZUhMq4= github.com/Scalingo/go-plugins-helpers v1.4.0 h1:TXysHoesEcH5Q4JBCo0prxWLFY0/yp2Q2WFBGa+Curo= github.com/Scalingo/go-plugins-helpers v1.4.0/go.mod h1:pPivLa4FeG8iB0nff7NwI7gIZ4bMr62u/prYXktoWIo= +github.com/Scalingo/go-utils/cronsetup v1.6.0 h1:mbaZXF1e1tTuZVb6D7nbwxFQVi+fjiEzByZ+vwnUfB8= +github.com/Scalingo/go-utils/cronsetup v1.6.0/go.mod h1:KOVhrti0nmP/szDhNAvdGpivJtXoNwXDfuU8Ywpqhyg= github.com/Scalingo/go-utils/crypto v1.1.1 h1:+F4JqC/8Lu8b1MDJwdQjT19yzLhShWPk+K93cUv7KVg= github.com/Scalingo/go-utils/crypto v1.1.1/go.mod h1:29dsjRaXxNI6tUG2ZbNWs4A5pqmwCAztdgfudGQDuhY= github.com/Scalingo/go-utils/errors/v2 v2.5.1 h1:1tfJW6/ZxTgrRmFTlKQCOtArQquOW0/XdZQzx8wMHoM= @@ -21,6 +23,8 @@ github.com/Scalingo/go-utils/graceful v1.3.3 h1:N1Gcw6nnu3Sid2icxE7G25c8JwKjBh4M github.com/Scalingo/go-utils/graceful v1.3.3/go.mod h1:d22EgSLL6g0uIeqQrsFAQbWqw5Qb5UPpgT0bTVf/e0s= github.com/Scalingo/go-utils/logger v1.12.2 h1:9vm83/gqjCIy5t+OuNYjkVOUrJtdMy78XNIv8E+OCCU= github.com/Scalingo/go-utils/logger v1.12.2/go.mod h1:vaeFcI5LMHiRRmMfJbbnblbj3RXRJIzxUcyEjZpMFpg= +github.com/Scalingo/go-utils/otel v0.8.0 h1:iLWJHjjpmYHmzly8XNsA5Kn89GpUaSk6Jie6yPRvtrc= +github.com/Scalingo/go-utils/otel v0.8.0/go.mod h1:ZNIXMGq+HCpUa3SGYZ5i5ivZavlXwVulDm75d6m+sd0= github.com/Scalingo/go-utils/security v1.2.2 h1:3AkvqQm0b58wsM4mtvh6vlaOEelRJuvI1iLviMTdXX4= github.com/Scalingo/go-utils/security v1.2.2/go.mod h1:oGgcjK4/PtG3rspMTygnZUl0z2cu/qS+JA+HGpKY0Lk= github.com/Scalingo/logrus-rollbar v1.4.4 h1:zi+0LJaHlDm9HcArOh8Gmqt0scNFC77wn4c9ePSNof0= @@ -60,6 +64,8 @@ github.com/gofrs/uuid/v5 v5.4.0 h1:EfbpCTjqMuGyq5ZJwxqzn3Cbr2d0rUZU7v5ycAk/e/0= github.com/gofrs/uuid/v5 v5.4.0/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc= +github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.2.1-0.20190312032427-6f77996f0c42/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= diff --git a/node/ensure.go b/node/ensure.go new file mode 100644 index 00000000..0136f26b --- /dev/null +++ b/node/ensure.go @@ -0,0 +1,88 @@ +package node + +import ( + "context" + "os" + + "github.com/pkg/errors" + "github.com/sirupsen/logrus" + + "github.com/Scalingo/go-utils/logger" + "github.com/Scalingo/sand/api/params" + "github.com/Scalingo/sand/config" + "github.com/Scalingo/sand/endpoint" + "github.com/Scalingo/sand/network" + "github.com/Scalingo/sand/store" +) + +func EnsureNetworkEndpoints(ctx context.Context, c *config.Config, repo network.Repository, erepo endpoint.Repository) error { + log := logger.Get(ctx) + ctx = logger.ToCtx(ctx, log) + + log.Info("Ensure networks on node") + + endpoints, err := erepo.List(ctx, map[string]string{"hostname": c.GetPeerHostname()}) + if errors.Cause(err) == store.ErrNotFound { + return nil + } + if err != nil { + return errors.Wrapf(err, "failed to list endpoints of %v", c.GetPeerHostname()) + } + + for _, endpoint := range endpoints { + endpointLog := log.WithFields(logrus.Fields{ + "endpoint_id": endpoint.ID, + }) + endpointCtx := logger.ToCtx(ctx, endpointLog) + + if !endpoint.Active { + endpointLog.Debug("skip inactive endpoint") + continue + } + + endpointLog = endpointLog.WithFields(logrus.Fields{ + "network_id": endpoint.NetworkID, + "endpoint_netns_path": endpoint.TargetNetnsPath, + }) + endpointCtx = logger.ToCtx(ctx, endpointLog) + endpointLog.Info("restoring endpoint") + + network, ok, err := repo.Exists(endpointCtx, endpoint.NetworkID) + if err != nil { + return errors.Wrapf(err, "failed to get network") + } + if !ok { + endpointLog.WithError(errors.Errorf("network not found for %v", endpoint)).Error("skip endpoint") + continue + } + + endpointLog.Info("ensuring network") + err = repo.Ensure(endpointCtx, network) + if err != nil { + endpointLog.WithError(err).Error("failed to ensure network") + continue + } + + _, err = erepo.Activate(endpointCtx, network, endpoint, params.EndpointActivate{ + NSHandlePath: endpoint.TargetNetnsPath, + SetAddr: true, + MoveVeth: true, + }) + if err != nil { + // If the netns path no longer exists, deactivate the endpoint. Other + // activation errors should not prevent the remaining endpoints from being ensured. + if os.IsNotExist(errors.Cause(err)) { + _, err = erepo.Deactivate(endpointCtx, network, endpoint) + if err != nil { + endpointLog.WithError(err).Error("failed to deactivate endpoint") + continue + } + } else { + endpointLog.WithError(err).Error("failed to ensure endpoint") + continue + } + } + } + + return nil +} diff --git a/node/ensure_test.go b/node/ensure_test.go new file mode 100644 index 00000000..19539468 --- /dev/null +++ b/node/ensure_test.go @@ -0,0 +1,104 @@ +package node + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + "github.com/Scalingo/sand/api/params" + "github.com/Scalingo/sand/api/types" + "github.com/Scalingo/sand/config" + "github.com/Scalingo/sand/test/mocks/endpointmock" + "github.com/Scalingo/sand/test/mocks/networkmock" +) + +func TestEnsureNetworkEndpoints(t *testing.T) { + cases := []struct { + Name string + ExpectNetworkRepository func(*networkmock.MockRepository) + ExpectEndpointRepository func(*endpointmock.MockRepository) + }{ + { + Name: "activate endpoints on current node", + ExpectEndpointRepository: func(r *endpointmock.MockRepository) { + endpoint := types.Endpoint{ + ID: "endpoint-1", + NetworkID: "network-1", + Active: true, + TargetNetnsPath: "/proc/1/ns/net", + } + network := types.Network{ID: "network-1"} + + r.EXPECT().List(gomock.Any(), map[string]string{"hostname": "node-1"}).Return([]types.Endpoint{endpoint}, nil) + r.EXPECT().Activate(gomock.Any(), network, endpoint, params.EndpointActivate{ + NSHandlePath: "/proc/1/ns/net", + SetAddr: true, + MoveVeth: true, + }).Return(endpoint, nil) + }, + ExpectNetworkRepository: func(r *networkmock.MockRepository) { + network := types.Network{ID: "network-1"} + + r.EXPECT().Exists(gomock.Any(), "network-1").Return(network, true, nil) + r.EXPECT().Ensure(gomock.Any(), network).Return(nil) + }, + }, { + Name: "skip inactive endpoints", + ExpectEndpointRepository: func(r *endpointmock.MockRepository) { + r.EXPECT().List(gomock.Any(), map[string]string{"hostname": "node-1"}).Return([]types.Endpoint{{ + ID: "endpoint-1", + NetworkID: "network-1", + Active: false, + }}, nil) + }, + }, { + Name: "deactivate endpoint when netns no longer exists", + ExpectEndpointRepository: func(r *endpointmock.MockRepository) { + endpoint := types.Endpoint{ + ID: "endpoint-1", + NetworkID: "network-1", + Active: true, + TargetNetnsPath: "/proc/1/ns/net", + } + network := types.Network{ID: "network-1"} + + r.EXPECT().List(gomock.Any(), map[string]string{"hostname": "node-1"}).Return([]types.Endpoint{endpoint}, nil) + r.EXPECT().Activate(gomock.Any(), network, endpoint, params.EndpointActivate{ + NSHandlePath: "/proc/1/ns/net", + SetAddr: true, + MoveVeth: true, + }).Return(types.Endpoint{}, os.ErrNotExist) + r.EXPECT().Deactivate(gomock.Any(), network, endpoint).Return(endpoint, nil) + }, + ExpectNetworkRepository: func(r *networkmock.MockRepository) { + network := types.Network{ID: "network-1"} + + r.EXPECT().Exists(gomock.Any(), "network-1").Return(network, true, nil) + r.EXPECT().Ensure(gomock.Any(), network).Return(nil) + }, + }, + } + + for _, c := range cases { + t.Run(c.Name, func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + networkRepo := networkmock.NewMockRepository(ctrl) + endpointRepo := endpointmock.NewMockRepository(ctrl) + + if c.ExpectNetworkRepository != nil { + c.ExpectNetworkRepository(networkRepo) + } + + if c.ExpectEndpointRepository != nil { + c.ExpectEndpointRepository(endpointRepo) + } + + err := EnsureNetworkEndpoints(t.Context(), &config.Config{PeerHostname: "node-1"}, networkRepo, endpointRepo) + require.NoError(t, err) + }) + } +} diff --git a/test/mocks/client/sandmock/client_mock.go b/test/mocks/client/sandmock/client_mock.go index cf864f89..121c5aec 100644 --- a/test/mocks/client/sandmock/client_mock.go +++ b/test/mocks/client/sandmock/client_mock.go @@ -149,6 +149,20 @@ func (mr *MockClientMockRecorder) NetworkShow(arg0, arg1 any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NetworkShow", reflect.TypeOf((*MockClient)(nil).NetworkShow), arg0, arg1) } +// NodeEnsureNetworkEndpoints mocks base method. +func (m *MockClient) NodeEnsureNetworkEndpoints(arg0 context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "NodeEnsureNetworkEndpoints", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// NodeEnsureNetworkEndpoints indicates an expected call of NodeEnsureNetworkEndpoints. +func (mr *MockClientMockRecorder) NodeEnsureNetworkEndpoints(arg0 any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NodeEnsureNetworkEndpoints", reflect.TypeOf((*MockClient)(nil).NodeEnsureNetworkEndpoints), arg0) +} + // NetworksList mocks base method. func (m *MockClient) NetworksList(arg0 context.Context) ([]types.Network, error) { m.ctrl.T.Helper() diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/CHANGELOG.md b/vendor/github.com/Scalingo/go-utils/cronsetup/CHANGELOG.md new file mode 100644 index 00000000..c253ca19 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/CHANGELOG.md @@ -0,0 +1,57 @@ +# Changelog + +## To be Released + +## v1.6.0 + +* feat(opentelemetry) Add ability to instrument execution of cron job through OpenTelemetry + +## v1.5.0 + +* feat(cronsetup): add local mode + +## v1.4.0 + +* feat(cronsetup): add option to provide etcd client +* refactor(cronsetup): replace `github.com/Scalingo/go-etcd-cron` with `cronsetup/internal/cron` + +## v1.3.0 + +* feat(request_id) Inject `request_id` in context for each cron execution + +## v1.2.1 + +* chore(go): corrective bump - Go version regression from 1.24.3 to 1.24 + +## v1.2.0 + +* chore(go): upgrade to Go 1.24 + +## v1.1.4 + +* Various dependencies updates + +## v1.1.3 + +* fix(cronsetup): add missing err check when adding job [#394](https://github.com/Scalingo/go-utils/pull/394) +* build(deps): bump github.com/Scalingo/go-utils/logger from 1.1.1 to 1.2.0 +* build(deps): bump go.etcd.io/etcd/client/v3 from 3.5.4 to 3.5.5 + +## v1.1.2 + +* chore(go): use go 1.17 +* build(deps): bump go.etcd.io/etcd/client/v3 from 3.5.0 to 3.5.4 + +## v1.1.1 + +* Bump github.com/go-utils/logger from v1.0.0 to v1.1.0 + +## v1.1.0 + +* Bump Scalingo/go-etcd-cron to 1.3.0 and bump etcd client to 3.5.0 + [#207](https://github.com/Scalingo/go-utils/pull/207) +* Bump go version to 1.16 + +## v1.0.0, v1.0.1, v1.0.2 + +* Initial breakdown of go-utils into subpackages diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/LICENSE b/vendor/github.com/Scalingo/go-utils/cronsetup/LICENSE new file mode 100644 index 00000000..8e1c1aa7 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2020 Scalingo + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/README.md b/vendor/github.com/Scalingo/go-utils/cronsetup/README.md new file mode 100644 index 00000000..0729330d --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/README.md @@ -0,0 +1,41 @@ +# Package `cronsetup` v1.6.0 + +The `cronsetup` package eases the configuration and the execution of cron jobs on Go services. It has two modes: a distributed one (backed by etcd), and a local one. + +## Goals + +### Distributed Mode + +The goal of the distributed mode is to implement a distributed and fault tolerant cron in order to: + +* Run an identical process on several hosts +* Each of these process instantiate a cron with the same rules +* Ensure only one of these processes executes an iteration of a job + +### Local Mode + +The goal of the local mode is to implement a classical cron: it executes each job on all hosts executing the cron task. + +## Examples + +Examples are available in the `examples` package. One can execute them with: + +```sh +go run ./examples/distributed +go run ./examples/local +``` + +The distributed example requires a etcd to be running at `127.0.0.1:2379`. + +## Telemetry + +Telemetry is enabled by default, and can be disabled with `WithoutTelemetry` set +to true in initialization option. + +It records the duration of each job execution (in seconds). The metric uses the +`scalingo.etcd_cron.job_name` attribute with the job name as value, and a +`scalingo.etcd_cron.status` attribute with `success` or `error`. + +Metrics: + +- `scalingo.etcd_cron.run.duration`: execution time of a job in seconds diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/cronsetup.go b/vendor/github.com/Scalingo/go-utils/cronsetup/cronsetup.go new file mode 100644 index 00000000..12e657b9 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/cronsetup.go @@ -0,0 +1,137 @@ +package cronsetup + +import ( + "context" + + "github.com/gofrs/uuid/v5" + "github.com/sirupsen/logrus" + etcdv3 "go.etcd.io/etcd/client/v3" + + "github.com/Scalingo/go-utils/cronsetup/internal/cron" + "github.com/Scalingo/go-utils/errors/v3" + "github.com/Scalingo/go-utils/logger" +) + +// Job represents a cron job. It contains 3 *mandatory* options to define a job. +type Job = cron.Job + +// SetupOpts are the options to setup new cron jobs. One of EtcdClient or EtcdConfig must be provided. +type SetupOpts struct { + // EtcdClient is the etcd client to use to set mutexes + EtcdClient *etcdv3.Client + // EtcdConfig is the configuration to use in order to create and etcd client to set mutexes + EtcdConfig func() (etcdv3.Config, error) + // List of jobs to execute + Jobs []Job + // WithoutTelemetry indicates whether OpenTelemetry instrumentation should be disabled + WithoutTelemetry bool +} + +// Setup configures a new etcd cron and starts it. The caller has the responsibility to call the returned function to stop the cron jobs. +// All errors returned by a cron job or by etcd are logged using the logger in the context. +func Setup(ctx context.Context, opts SetupOpts) (func(), error) { + log := logger.Get(ctx) + + if opts.EtcdClient != nil && opts.EtcdConfig != nil { + return nil, errors.New(ctx, "both etcd client and config cannot be set") + } + + cronOpts := []cron.Opt{ + cron.WithFuncCtx(funcCtx), + cron.WithErrorsHandler(errorHandler), + cron.WithEtcdErrorsHandler(errorHandler), + } + + if opts.EtcdClient == nil && opts.EtcdConfig == nil { + ctx, log = logger.WithFieldToCtx(ctx, "mode", "local") + } else { + ctx, log = logger.WithFieldToCtx(ctx, "mode", "distributed") + + etcdMutexBuilder, err := createEtcdMutextBuilderFromOpts(ctx, opts) + if err != nil { + return nil, errors.Wrap(ctx, err, "create the etcd mutex builder") + } + + cronOpts = append(cronOpts, cron.WithEtcdMutexBuilder(etcdMutexBuilder)) + } + + c, err := cron.New(cronOpts...) + if err != nil { + return nil, errors.Wrap(ctx, err, "create cron job runner") + } + + var telemetry *telemetry + if !opts.WithoutTelemetry { + telemetry, err = newTelemetry(ctx) + if err != nil { + return nil, errors.Wrap(ctx, err, "init telemetry") + } + } + + for _, job := range opts.Jobs { + if telemetry != nil { + job = telemetry.wrapJob(job) + } + + err := c.AddJob(job) + if err != nil { + return nil, errors.Wrap(ctx, err, "add the cron job") + } + } + + log.Info("Starting cron goroutine") + + c.Start(ctx) + return func() { + log.Info("Stopping cron goroutine") + c.Stop() + }, nil +} + +func createEtcdMutextBuilderFromOpts(ctx context.Context, opts SetupOpts) (cron.EtcdMutexBuilder, error) { + if opts.EtcdClient != nil { + etcdMutexBuilder, err := cron.NewEtcdMutexBuilderFromClient(opts.EtcdClient) + if err != nil { + return nil, errors.Wrap(ctx, err, "create etcd mutex builder from client") + } + return etcdMutexBuilder, nil + } + + etcdConfig, err := opts.EtcdConfig() + if err != nil { + return nil, errors.Wrap(ctx, err, "get etcd config") + } + + etcdMutexBuilder, err := cron.NewEtcdMutexBuilder(etcdConfig) + if err != nil { + return nil, errors.Wrap(ctx, err, "create etcd mutex builder from config") + } + + return etcdMutexBuilder, nil +} + +func funcCtx(ctx context.Context, j cron.Job) context.Context { + log := logger.Get(ctx) + requestID, ok := ctx.Value("request_id").(string) + if !ok { + requestUUID, err := uuid.NewV4() + if err != nil { + log.WithError(err).Error("Error generating UUID v4") + } else { + requestID = requestUUID.String() + //nolint:revive,staticcheck // The "request_id" should not be of type string (https://pkg.go.dev/context#WithValue). + // I don't know what would be the impact of using another type for such a field that is used in various repositories. Hence I'm disabling the linters. + ctx = context.WithValue(ctx, "request_id", requestID) + } + } + ctx, _ = logger.WithFieldsToCtx(ctx, logrus.Fields{ + "job_name": j.Name, + "request_id": requestID, + }) + return ctx +} + +func errorHandler(ctx context.Context, _ cron.Job, err error) { + log := logger.Get(ctx) + log.WithError(err).Error("Error when running cron job") +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/LICENSE b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/LICENSE new file mode 100644 index 00000000..3a0f627f --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/LICENSE @@ -0,0 +1,21 @@ +Copyright (C) 2012 Rob Figueiredo +All Rights Reserved. + +MIT LICENSE + +Permission is hereby granted, free of charge, to any person obtaining a copy of +this software and associated documentation files (the "Software"), to deal in +the Software without restriction, including without limitation the rights to +use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/constantdelay.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/constantdelay.go new file mode 100644 index 00000000..cd6e7b1b --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/constantdelay.go @@ -0,0 +1,27 @@ +package cron + +import "time" + +// ConstantDelaySchedule represents a simple recurring duty cycle, e.g. "Every 5 minutes". +// It does not support jobs more frequent than once a second. +type ConstantDelaySchedule struct { + Delay time.Duration +} + +// Every returns a crontab Schedule that activates once every duration. +// Delays of less than a second are not supported (will round up to 1 second). +// Any fields less than a Second are truncated. +func Every(duration time.Duration) ConstantDelaySchedule { + if duration < time.Second { + duration = time.Second + } + return ConstantDelaySchedule{ + Delay: duration - time.Duration(duration.Nanoseconds())%time.Second, + } +} + +// Next returns the next time this should be run. +// This rounds so that the next activation time will be on the second. +func (schedule ConstantDelaySchedule) Next(t time.Time) time.Time { + return t.Add(schedule.Delay - time.Duration(t.Nanosecond())*time.Nanosecond) +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/cron.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/cron.go new file mode 100644 index 00000000..cbe69c99 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/cron.go @@ -0,0 +1,375 @@ +package cron + +import ( + "context" + "fmt" + "regexp" + "runtime/debug" + "sort" + "strings" + "time" + + "github.com/Scalingo/go-utils/errors/v3" + "github.com/Scalingo/go-utils/logger" +) + +// Cron keeps track of any number of entries, invoking the associated func as +// specified by the schedule. It may be started, stopped, and the entries may +// be inspected while running. +type Cron struct { + entries []*Entry + stop chan struct{} + add chan *Entry + snapshot chan []*Entry + etcdErrorsHandler func(context.Context, Job, error) + errorsHandler func(context.Context, Job, error) + funcCtx func(context.Context, Job) context.Context + running bool + etcdMutextBuilder EtcdMutexBuilder +} + +// Job contains 3 mandatory options to define a job +type Job struct { + // Name of the job + Name string + // Cron-formatted rhythm (ie. 0,10,30 1-5 0 * * *) + Rhythm string + // Routine method + Func func(context.Context) error +} + +func (j Job) Run(ctx context.Context) error { + return j.Func(ctx) +} + +var ( + nonAlphaNumerical = regexp.MustCompile("[^a-z0-9_]") +) + +func (j Job) canonicalName() string { + jobNameLowerCase := strings.ToLower(j.Name) + // Replace non alphanumeric characters with '_' + jobNameWithoutSpecialCharacters := nonAlphaNumerical.ReplaceAllString( + jobNameLowerCase, "_", + ) + return toSnakeCase(jobNameWithoutSpecialCharacters) +} + +// toSnakeCase converts the string s to a snake case: +// - all spaces are replaced with a _ +// - non-alphanumeric characters are replaced with a _ +// +// Credits goes to https://github.com/iancoleman/strcase under the MIT license for this code. +func toSnakeCase(s string) string { + delimiter := byte('_') + s = strings.TrimSpace(s) + n := strings.Builder{} + n.Grow(len(s) + 2) // nominal 2 bytes of extra space for inserted delimiters + for i, v := range []byte(s) { + vIsCap := v >= 'A' && v <= 'Z' + vIsLow := v >= 'a' && v <= 'z' + if vIsCap { + v += 'a' + v -= 'A' + } + + // treat acronyms as words, eg for JSONData -> JSON is a whole word + //nolint:nestif + if i+1 < len(s) { + next := s[i+1] + vIsNum := v >= '0' && v <= '9' + nextIsCap := next >= 'A' && next <= 'Z' + nextIsLow := next >= 'a' && next <= 'z' + nextIsNum := next >= '0' && next <= '9' + // add underscore if next letter case type is changed + if (vIsCap && (nextIsLow || nextIsNum)) || (vIsLow && (nextIsCap || nextIsNum)) || (vIsNum && (nextIsCap || nextIsLow)) { + if vIsCap && nextIsLow { + if prevIsCap := i > 0 && s[i-1] >= 'A' && s[i-1] <= 'Z'; prevIsCap { + n.WriteByte(delimiter) + } + } + n.WriteByte(v) + if vIsLow || vIsNum || nextIsNum { + n.WriteByte(delimiter) + } + continue + } + } + + if v == ' ' || v == '_' || v == '-' || v == '.' { + n.WriteByte(delimiter) + } else { + n.WriteByte(v) + } + } + + return n.String() +} + +// The Schedule describes a job's duty cycle. +type Schedule interface { + // Return the next activation time, later than the given time. + // Next is invoked initially, and then each time the job is run. + Next(t time.Time) time.Time +} + +// Entry consists of a schedule and the func to execute on that schedule. +type Entry struct { + // The schedule on which this job should be run. + Schedule Schedule + + // The next time the job will run. This is the zero time if Cron has not been + // started or this entry's schedule is unsatisfiable + Next time.Time + + // The last time this job was run. This is the zero time if the job has never + // been run. + Prev time.Time + + // The Job o run. + Job Job +} + +// byTime is a wrapper for sorting the entry array by time +// (with zero time at the end). +type byTime []*Entry + +func (s byTime) Len() int { return len(s) } +func (s byTime) Swap(i, j int) { s[i], s[j] = s[j], s[i] } +func (s byTime) Less(i, j int) bool { + // Two zero times should return false. + // Otherwise, zero is "greater" than any other time. + // (To sort it at the end of the list.) + if s[i].Next.IsZero() { + return false + } + if s[j].Next.IsZero() { + return true + } + return s[i].Next.Before(s[j].Next) +} + +type Opt func(cron *Cron) + +// WithEtcdErrorsHandler updates the default etcd error handler. It is called when an error occurs while interacting with etcd. +// The default handler outputs a log line on stdout. +func WithEtcdErrorsHandler(f func(context.Context, Job, error)) Opt { + return Opt(func(cron *Cron) { + cron.etcdErrorsHandler = f + }) +} + +// WithErrorsHandler updates the default error handler. It is called when an error occurs while executing a cron job. +// The default handler outputs a log line on stdout. +func WithErrorsHandler(f func(context.Context, Job, error)) Opt { + return Opt(func(cron *Cron) { + cron.errorsHandler = f + }) +} + +// WithEtcdMutexBuilder sets an etcd client to pose mutex. Setting such a client enables the distributed mode. +func WithEtcdMutexBuilder(etcdMutexBuilder EtcdMutexBuilder) Opt { + return Opt(func(cron *Cron) { + cron.etcdMutextBuilder = etcdMutexBuilder + }) +} + +// WithFuncCtx is a callback executed at the beginning of the execution of each entry. It only returns a context. +func WithFuncCtx(f func(context.Context, Job) context.Context) Opt { + return Opt(func(cron *Cron) { + cron.funcCtx = f + }) +} + +// New returns a new cron job runner. +func New(opts ...Opt) (*Cron, error) { + cron := &Cron{ + entries: nil, + add: make(chan *Entry), + stop: make(chan struct{}), + snapshot: make(chan []*Entry), + running: false, + } + for _, opt := range opts { + opt(cron) + } + + if cron.etcdErrorsHandler == nil { + cron.etcdErrorsHandler = func(ctx context.Context, j Job, err error) { + _, log := logger.WithFieldToCtx(ctx, "job_name", j.Name) + log.WithError(err).Infof("[cron] etcd error when handling '%v' job: %v", j.Name, err) + } + } + + if cron.errorsHandler == nil { + cron.errorsHandler = func(ctx context.Context, j Job, err error) { + _, log := logger.WithFieldToCtx(ctx, "job_name", j.Name) + log.WithError(err).Infof("[cron] error when handling '%v' job: %v", j.Name, err) + } + } + + return cron, nil +} + +// AddFunc adds a Job to the Cron to be run on the given schedule. +func (c *Cron) AddJob(job Job) error { + schedule, err := Parse(job.Rhythm) + if err != nil { + return err + } + c.Schedule(schedule, job) + return nil +} + +// Schedule adds a Job to the Cron to be run on the given schedule. +func (c *Cron) Schedule(schedule Schedule, job Job) { + entry := &Entry{ + Schedule: schedule, + Job: job, + } + if !c.running { + c.entries = append(c.entries, entry) + return + } + + c.add <- entry +} + +// Entries returns a snapshot of the cron entries. +func (c *Cron) Entries() []*Entry { + if c.running { + c.snapshot <- nil + x := <-c.snapshot + return x + } + return c.entrySnapshot() +} + +// Start the cron scheduler in its own go-routine. +func (c *Cron) Start(ctx context.Context) { + c.running = true + go c.run(ctx) +} + +// Run the scheduler. This is private just due to the need to synchronize +// access to the 'running' state variable. +func (c *Cron) run(ctx context.Context) { + // Figure out the next activation times for each entry. + now := time.Now().Local() + for _, entry := range c.entries { + entry.Next = entry.Schedule.Next(now) + } + + for { + // Determine the next entry to run. + sort.Sort(byTime(c.entries)) + + var effective time.Time + if len(c.entries) == 0 || c.entries[0].Next.IsZero() { + // If there are no entries yet, just sleep - it still handles new entries + // and stop requests. + effective = now.AddDate(10, 0, 0) + } else { + effective = c.entries[0].Next + } + + select { + case now = <-time.After(effective.Sub(now)): + // Run every entry whose next time was this effective time. + for _, e := range c.entries { + if e.Next != effective { + break + } + e.Prev = e.Next + e.Next = e.Schedule.Next(effective) + + go c.runEntry(ctx, effective, e) + } + continue + + case newEntry := <-c.add: + c.entries = append(c.entries, newEntry) + newEntry.Next = newEntry.Schedule.Next(now) + + case <-c.snapshot: + c.snapshot <- c.entrySnapshot() + + case <-c.stop: + return + } + + // 'now' should be updated after newEntry and snapshot cases. + now = time.Now().Local() + } +} + +// Stop the cron scheduler. +func (c *Cron) Stop() { + c.stop <- struct{}{} + c.running = false +} + +// entrySnapshot returns a copy of the current cron entry list. +func (c *Cron) entrySnapshot() []*Entry { + entries := []*Entry{} + for _, e := range c.entries { + entries = append(entries, &Entry{ + Schedule: e.Schedule, + Next: e.Next, + Prev: e.Prev, + Job: e.Job, + }) + } + return entries +} + +func (c *Cron) runEntry(ctx context.Context, effective time.Time, e *Entry) { + defer func() { + r := recover() + if r != nil { + err, ok := r.(error) + if !ok { + err = fmt.Errorf("%v", r) + } + err = fmt.Errorf("panic: %v, stacktrace: %s", err, string(debug.Stack())) + go c.errorsHandler(ctx, e.Job, err) + } + }() + + if c.funcCtx != nil { + ctx = c.funcCtx(ctx, e.Job) + } + + if c.etcdMutextBuilder == nil { + // In the local mode, we execute the job anyway with no need of any mutex + err := e.Job.Run(ctx) + if err != nil { + go c.errorsHandler(ctx, e.Job, err) + } + + return + } + + // In the distributed mode, we need to set a distributed mutex to ensure the job is only executed once. + m, err := c.etcdMutextBuilder.NewMutex(fmt.Sprintf("etcd_cron/%s/%d", e.Job.canonicalName(), effective.Unix())) + if err != nil { + go c.etcdErrorsHandler(ctx, e.Job, errors.Wrapf(ctx, err, "create etcd mutex for job '%v'", e.Job.Name)) + return + } + lockCtx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + + err = m.Lock(lockCtx) + if err == context.DeadlineExceeded { + return + } else if err != nil { + go c.etcdErrorsHandler(ctx, e.Job, errors.Wrapf(ctx, err, "lock mutex '%v'", m.Key())) + return + } + + err = e.Job.Run(ctx) + if err != nil { + go c.errorsHandler(ctx, e.Job, err) + return + } +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/doc.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/doc.go new file mode 100644 index 00000000..47fbf9b4 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/doc.go @@ -0,0 +1,129 @@ +/* +Package cron implements a cron spec parser and job runner. + +# Usage + +Callers may register Funcs to be invoked on a given schedule. Cron will run +them in their own goroutines. + + c := cron.New() + c.AddFunc("0 30 * * * *", func() { fmt.Println("Every hour on the half hour") }) + c.AddFunc("@hourly", func() { fmt.Println("Every hour") }) + c.AddFunc("@every 1h30m", func() { fmt.Println("Every hour thirty") }) + c.Start() + .. + // Funcs are invoked in their own goroutine, asynchronously. + ... + // Funcs may also be added to a running Cron + c.AddFunc("@daily", func() { fmt.Println("Every day") }) + .. + // Inspect the cron job entries' next and previous run times. + inspect(c.Entries()) + .. + c.Stop() // Stop the scheduler (does not stop any jobs already running). + +# CRON Expression Format + +A cron expression represents a set of times, using 6 space-separated fields. + + Field name | Mandatory? | Allowed values | Allowed special characters + ---------- | ---------- | -------------- | -------------------------- + Seconds | Yes | 0-59 | * / , - + Minutes | Yes | 0-59 | * / , - + Hours | Yes | 0-23 | * / , - + Day of month | Yes | 1-31 | * / , - ? + Month | Yes | 1-12 or JAN-DEC | * / , - + Day of week | Yes | 0-6 or SUN-SAT | * / , - ? + +Note: Month and Day-of-week field values are case insensitive. "SUN", "Sun", +and "sun" are equally accepted. + +# Special Characters + +Asterisk ( * ) + +The asterisk indicates that the cron expression will match for all values of the +field; e.g., using an asterisk in the 5th field (month) would indicate every +month. + +Slash ( / ) + +Slashes are used to describe increments of ranges. For example 3-59/15 in the +1st field (minutes) would indicate the 3rd minute of the hour and every 15 +minutes thereafter. The form "*\/..." is equivalent to the form "first-last/...", +that is, an increment over the largest possible range of the field. The form +"N/..." is accepted as meaning "N-MAX/...", that is, starting at N, use the +increment until the end of that specific range. It does not wrap around. + +Comma ( , ) + +Commas are used to separate items of a list. For example, using "MON,WED,FRI" in +the 5th field (day of week) would mean Mondays, Wednesdays and Fridays. + +Hyphen ( - ) + +Hyphens are used to define ranges. For example, 9-17 would indicate every +hour between 9am and 5pm inclusive. + +Question mark ( ? ) + +Question mark may be used instead of '*' for leaving either day-of-month or +day-of-week blank. + +# Predefined schedules + +You may use one of several pre-defined schedules in place of a cron expression. + + Entry | Description | Equivalent To + ----- | ----------- | ------------- + @yearly (or @annually) | Run once a year, midnight, Jan. 1st | 0 0 0 1 1 * + @monthly | Run once a month, midnight, first of month | 0 0 0 1 * * + @weekly | Run once a week, midnight on Sunday | 0 0 0 * * 0 + @daily (or @midnight) | Run once a day, midnight | 0 0 0 * * * + @hourly | Run once an hour, beginning of hour | 0 0 * * * * + +# Intervals + +You may also schedule a job to execute at fixed intervals. This is supported by +formatting the cron spec like this: + + @every + +where "duration" is a string accepted by time.ParseDuration +(http://golang.org/pkg/time/#ParseDuration). + +For example, "@every 1h30m10s" would indicate a schedule that activates every +1 hour, 30 minutes, 10 seconds. + +Note: The interval does not take the job runtime into account. For example, +if a job takes 3 minutes to run, and it is scheduled to run every 5 minutes, +it will have only 2 minutes of idle time between each run. + +# Time zones + +All interpretation and scheduling is done in the machine's local time zone (as +provided by the Go time package (http://www.golang.org/pkg/time). + +Be aware that jobs scheduled during daylight-savings leap-ahead transitions will +not be run! + +# Thread safety + +Since the Cron service runs concurrently with the calling code, some amount of +care must be taken to ensure proper synchronization. + +All cron methods are designed to be correctly synchronized as long as the caller +ensures that invocations have a clear happens-before ordering between them. + +# Implementation + +Cron entries are stored in an array, sorted by their next activation time. Cron +sleeps until the next job is due to be run. + +Upon waking: + - it runs each entry that is active on that second + - it calculates the next run times for the jobs that were run + - it re-sorts the array of entries by next activation time. + - it goes to sleep until the soonest job. +*/ +package cron diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/etcd.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/etcd.go new file mode 100644 index 00000000..dcfcf6cd --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/etcd.go @@ -0,0 +1,47 @@ +package cron + +import ( + "context" + + etcdv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/client/v3/concurrency" +) + +type DistributedMutex interface { + IsOwner() etcdv3.Cmp + Key() string + Lock(ctx context.Context) error + Unlock(ctx context.Context) error +} + +type EtcdMutexBuilder interface { + NewMutex(pfx string) (DistributedMutex, error) +} + +type etcdMutexBuilder struct { + *etcdv3.Client +} + +func NewEtcdMutexBuilderFromClient(c *etcdv3.Client) (EtcdMutexBuilder, error) { + return etcdMutexBuilder{Client: c}, nil +} + +func NewEtcdMutexBuilder(config etcdv3.Config) (EtcdMutexBuilder, error) { + c, err := etcdv3.New(config) + if err != nil { + return nil, err + } + return etcdMutexBuilder{Client: c}, nil +} + +func (c etcdMutexBuilder) NewMutex(pfx string) (DistributedMutex, error) { + // As each task iteration lock name is unique, we don't really care about unlocking it + // So the etcd lease will last 10 minutes, it ensures that even if another server + // clock is ill-configured (with a maximum span of 10 minutes), it won't execute the task + // twice. + session, err := concurrency.NewSession(c.Client, concurrency.WithTTL(60*10)) + if err != nil { + return nil, err + } + return concurrency.NewMutex(session, pfx), nil +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/parser.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/parser.go new file mode 100644 index 00000000..2787cc6e --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/parser.go @@ -0,0 +1,231 @@ +package cron + +import ( + "fmt" + "log" + "math" + "strconv" + "strings" + "time" +) + +// Parse returns a new crontab schedule representing the given spec. +// It returns a descriptive error if the spec is not valid. +// +// It accepts +// - Full crontab specs, e.g. "* * * * * ?" +// - Descriptors, e.g. "@midnight", "@every 1h30m" +func Parse(spec string) (_ Schedule, err error) { + // Convert panics into errors + defer func() { + if recovered := recover(); recovered != nil { + err = fmt.Errorf("%v", recovered) + } + }() + + if spec[0] == '@' { + return parseDescriptor(spec), nil + } + + // Split on whitespace. We require 5 or 6 fields. + // (second) (minute) (hour) (day of month) (month) (day of week, optional) + fields := strings.Fields(spec) + if len(fields) != 5 && len(fields) != 6 { + log.Panicf("Expected 5 or 6 fields, found %d: %s", len(fields), spec) + } + + // If a sixth field is not provided (DayOfWeek), then it is equivalent to star. + if len(fields) == 5 { + fields = append(fields, "*") + } + + schedule := &SpecSchedule{ + Second: getField(fields[0], seconds), + Minute: getField(fields[1], minutes), + Hour: getField(fields[2], hours), + Dom: getField(fields[3], dom), + Month: getField(fields[4], months), + Dow: getField(fields[5], dow), + } + + return schedule, nil +} + +// getField returns an Int with the bits set representing all of the times that +// the field represents. A "field" is a comma-separated list of "ranges". +func getField(field string, r bounds) uint64 { + // list = range {"," range} + var bits uint64 + ranges := strings.FieldsFunc(field, func(r rune) bool { return r == ',' }) + for _, expr := range ranges { + bits |= getRange(expr, r) + } + return bits +} + +// getRange returns the bits indicated by the given expression: +// +// number | number "-" number [ "/" number ] +func getRange(expr string, r bounds) uint64 { + var ( + start, end, step uint + rangeAndStep = strings.Split(expr, "/") + lowAndHigh = strings.Split(rangeAndStep[0], "-") + singleDigit = len(lowAndHigh) == 1 + ) + + var extraStar uint64 + if lowAndHigh[0] == "*" || lowAndHigh[0] == "?" { + start = r.min + end = r.max + extraStar = starBit + } else { + start = parseIntOrName(lowAndHigh[0], r.names) + switch len(lowAndHigh) { + case 1: + end = start + case 2: + end = parseIntOrName(lowAndHigh[1], r.names) + default: + log.Panicf("Too many hyphens: %s", expr) + } + } + + switch len(rangeAndStep) { + case 1: + step = 1 + case 2: + step = mustParseInt(rangeAndStep[1]) + + // Special handling: "N/step" means "N-max/step". + if singleDigit { + end = r.max + } + default: + log.Panicf("Too many slashes: %s", expr) + } + + if start < r.min { + log.Panicf("Beginning of range (%d) below minimum (%d): %s", start, r.min, expr) + } + if end > r.max { + log.Panicf("End of range (%d) above maximum (%d): %s", end, r.max, expr) + } + if start > end { + log.Panicf("Beginning of range (%d) beyond end of range (%d): %s", start, end, expr) + } + + return getBits(start, end, step) | extraStar +} + +// parseIntOrName returns the (possibly-named) integer contained in expr. +func parseIntOrName(expr string, names map[string]uint) uint { + if names != nil { + if namedInt, ok := names[strings.ToLower(expr)]; ok { + return namedInt + } + } + return mustParseInt(expr) +} + +// mustParseInt parses the given expression as an int or panics. +func mustParseInt(expr string) uint { + num, err := strconv.Atoi(expr) + if err != nil { + log.Panicf("Failed to parse int from %s: %s", expr, err) + } + if num < 0 { + log.Panicf("Negative number (%d) not allowed: %s", num, expr) + } + + return uint(num) +} + +// getBits sets all bits in the range [minVal, maxVal], modulo the given step size. +func getBits(minVal, maxVal, step uint) uint64 { + var bits uint64 + + // If step is 1, use shifts. + if step == 1 { + return ^(math.MaxUint64 << (maxVal + 1)) & (math.MaxUint64 << minVal) + } + + // Else, use a simple loop. + for i := minVal; i <= maxVal; i += step { + bits |= 1 << i + } + return bits +} + +// all returns all bits within the given bounds. (plus the star bit) +func all(r bounds) uint64 { + return getBits(r.min, r.max, 1) | starBit +} + +// parseDescriptor returns a pre-defined schedule for the expression, or panics +// if none matches. +func parseDescriptor(spec string) Schedule { + switch spec { + case "@yearly", "@annually": + return &SpecSchedule{ + Second: 1 << seconds.min, + Minute: 1 << minutes.min, + Hour: 1 << hours.min, + Dom: 1 << dom.min, + Month: 1 << months.min, + Dow: all(dow), + } + + case "@monthly": + return &SpecSchedule{ + Second: 1 << seconds.min, + Minute: 1 << minutes.min, + Hour: 1 << hours.min, + Dom: 1 << dom.min, + Month: all(months), + Dow: all(dow), + } + + case "@weekly": + return &SpecSchedule{ + Second: 1 << seconds.min, + Minute: 1 << minutes.min, + Hour: 1 << hours.min, + Dom: all(dom), + Month: all(months), + Dow: 1 << dow.min, + } + + case "@daily", "@midnight": + return &SpecSchedule{ + Second: 1 << seconds.min, + Minute: 1 << minutes.min, + Hour: 1 << hours.min, + Dom: all(dom), + Month: all(months), + Dow: all(dow), + } + + case "@hourly": + return &SpecSchedule{ + Second: 1 << seconds.min, + Minute: 1 << minutes.min, + Hour: all(hours), + Dom: all(dom), + Month: all(months), + Dow: all(dow), + } + } + + const every = "@every " + if strings.HasPrefix(spec, every) { + duration, err := time.ParseDuration(spec[len(every):]) + if err != nil { + log.Panicf("Failed to parse duration %s: %s", spec, err) + } + return Every(duration) + } + + log.Panicf("Unrecognized descriptor: %s", spec) + return nil +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/spec.go b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/spec.go new file mode 100644 index 00000000..8b672cac --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/internal/cron/spec.go @@ -0,0 +1,162 @@ +package cron + +import ( + "time" +) + +// SpecSchedule specifies a duty cycle (to the second granularity), based on a +// traditional crontab specification. It is computed initially and stored as bit sets. +type SpecSchedule struct { + Second, Minute, Hour, Dom, Month, Dow uint64 +} + +// bounds provides a range of acceptable values (plus a map of name to value). +type bounds struct { + min, max uint + names map[string]uint +} + +// The bounds for each field. +var ( + seconds = bounds{0, 59, nil} + minutes = bounds{0, 59, nil} + hours = bounds{0, 23, nil} + dom = bounds{1, 31, nil} + months = bounds{1, 12, map[string]uint{ + "jan": 1, + "feb": 2, + "mar": 3, + "apr": 4, + "may": 5, + "jun": 6, + "jul": 7, + "aug": 8, + "sep": 9, + "oct": 10, + "nov": 11, + "dec": 12, + }} + dow = bounds{0, 6, map[string]uint{ + "sun": 0, + "mon": 1, + "tue": 2, + "wed": 3, + "thu": 4, + "fri": 5, + "sat": 6, + }} +) + +const ( + // Set the top bit if a star was included in the expression. + starBit = 1 << 63 +) + +// Next returns the next time this schedule is activated, greater than the given +// time. If no time can be found to satisfy the schedule, return the zero time. +func (s *SpecSchedule) Next(t time.Time) time.Time { + // General approach: + // For Month, Day, Hour, Minute, Second: + // Check if the time value matches. If yes, continue to the next field. + // If the field doesn't match the schedule, then increment the field until it matches. + // While incrementing the field, a wrap-around brings it back to the beginning + // of the field list (since it is necessary to re-verify previous field + // values) + + // Start at the earliest possible time (the upcoming second). + t = t.Add(1*time.Second - time.Duration(t.Nanosecond())*time.Nanosecond) + + // This flag indicates whether a field has been incremented. + added := false + + // If no time is found within five years, return zero. + yearLimit := t.Year() + 5 + +WRAP: + if t.Year() > yearLimit { + return time.Time{} + } + + // Find the first applicable month. + // If it's this month, then do nothing. + for 1< 0 + dowMatch = 1< 0 + ) + + if s.Dom&starBit > 0 || s.Dow&starBit > 0 { + return domMatch && dowMatch + } + return domMatch || dowMatch +} diff --git a/vendor/github.com/Scalingo/go-utils/cronsetup/telemetry.go b/vendor/github.com/Scalingo/go-utils/cronsetup/telemetry.go new file mode 100644 index 00000000..b530f266 --- /dev/null +++ b/vendor/github.com/Scalingo/go-utils/cronsetup/telemetry.go @@ -0,0 +1,68 @@ +package cronsetup + +import ( + "context" + "time" + + otelsdk "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + "github.com/Scalingo/go-utils/cronsetup/internal/cron" + "github.com/Scalingo/go-utils/errors/v3" +) + +type telemetry struct { + runsDuration metric.Float64Histogram +} + +// jobNameAttributeKey captures the executed job name. +const jobNameAttributeKey = "scalingo.etcd_cron.job_name" + +// statusAttributeKey captures the execution status. +const statusAttributeKey = "scalingo.etcd_cron.status" + +const ( + statusSuccess = "success" + statusError = "error" +) + +func newTelemetry(ctx context.Context) (*telemetry, error) { + meter := otelsdk.Meter("scalingo.etcd_cron") + + runsDuration, err := meter.Float64Histogram( + "scalingo.etcd_cron.run.duration", + metric.WithDescription("Cron job execution duration in seconds"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, errors.Wrap(ctx, err, "create runs duration histogram") + } + + return &telemetry{ + runsDuration: runsDuration, + }, nil +} + +func (t *telemetry) wrapJob(job cron.Job) cron.Job { + originalFunc := job.Func + + job.Func = func(ctx context.Context) error { + startedAt := time.Now() + err := originalFunc(ctx) + + status := statusSuccess + if err != nil { + status = statusError + } + attributes := metric.WithAttributes( + attribute.String(jobNameAttributeKey, job.Name), + attribute.String(statusAttributeKey, status), + ) + t.runsDuration.Record(ctx, time.Since(startedAt).Seconds(), attributes) + + return err + } + + return job +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 9dac55a3..063211fb 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -19,6 +19,10 @@ github.com/Scalingo/go-handlers github.com/Scalingo/go-plugins-helpers/ipam github.com/Scalingo/go-plugins-helpers/network github.com/Scalingo/go-plugins-helpers/sdk +# github.com/Scalingo/go-utils/cronsetup v1.6.0 +## explicit; go 1.24.0 +github.com/Scalingo/go-utils/cronsetup +github.com/Scalingo/go-utils/cronsetup/internal/cron # github.com/Scalingo/go-utils/crypto v1.1.1 ## explicit; go 1.24 github.com/Scalingo/go-utils/crypto diff --git a/web/networks_controller_ensure.go b/web/networks_controller_ensure.go new file mode 100644 index 00000000..0469e40b --- /dev/null +++ b/web/networks_controller_ensure.go @@ -0,0 +1,21 @@ +package web + +import ( + "net/http" + + "github.com/pkg/errors" + + "github.com/Scalingo/sand/node" +) + +func (c NetworksController) EnsureNetworkEndpoints(w http.ResponseWriter, r *http.Request, params map[string]string) error { + w.Header().Set("Content-Type", "application/json") + + err := node.EnsureNetworkEndpoints(r.Context(), c.Config, c.NetworkRepository, c.EndpointRepository) + if err != nil { + return errors.Wrap(err, "ensure network endpoints") + } + + w.WriteHeader(http.StatusNoContent) + return nil +}