diff --git a/cluster/calcium/calcium_test.go b/cluster/calcium/calcium_test.go index fa25fd1cc..f64b6afcf 100644 --- a/cluster/calcium/calcium_test.go +++ b/cluster/calcium/calcium_test.go @@ -13,6 +13,7 @@ import ( "github.com/stretchr/testify/mock" "golang.org/x/crypto/ssh" + lockmocks "github.com/projecteru2/core/lock/mocks" "github.com/projecteru2/core/metrics" resourcemocks "github.com/projecteru2/core/resource/mocks" sourcemocks "github.com/projecteru2/core/source/mocks" @@ -110,3 +111,18 @@ func NewTestCluster() *Calcium { return c } + +func heldLock(t *testing.T) *lockmocks.DistributedLock { + t.Helper() + lock := &lockmocks.DistributedLock{} + lock.On("Lock", mock.Anything).Return(t.Context(), nil) + lock.On("Unlock", mock.Anything).Return(nil) + return lock +} + +func mockLocks(t *testing.T, store *storemocks.Store) *lockmocks.DistributedLock { + t.Helper() + lock := heldLock(t) + store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + return lock +} diff --git a/cluster/calcium/capacity_test.go b/cluster/calcium/capacity_test.go index 1c5d6f3f4..473467eab 100644 --- a/cluster/calcium/capacity_test.go +++ b/cluster/calcium/capacity_test.go @@ -1,14 +1,12 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" plugintypes "github.com/projecteru2/core/resource/plugins/types" resourcetypes "github.com/projecteru2/core/resource/types" @@ -22,10 +20,7 @@ func TestCalculateCapacity(t *testing.T) { ctx := t.Context() store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) engine := &enginemocks.API{} name := "n1" diff --git a/cluster/calcium/control_test.go b/cluster/calcium/control_test.go index 1824eb604..e5b84573a 100644 --- a/cluster/calcium/control_test.go +++ b/cluster/calcium/control_test.go @@ -11,7 +11,6 @@ import ( "github.com/projecteru2/core/cluster" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -212,9 +211,6 @@ func newControlTestCluster(t *testing.T) (*Calcium, context.Context, *storemocks c := NewTestCluster() ctx := t.Context() store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) return c, ctx, store } diff --git a/cluster/calcium/copy_test.go b/cluster/calcium/copy_test.go index 779c13437..15caa8d1e 100644 --- a/cluster/calcium/copy_test.go +++ b/cluster/calcium/copy_test.go @@ -1,14 +1,12 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -31,10 +29,7 @@ func TestCopy(t *testing.T) { }, } store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetWorkload", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() ch, err := c.Copy(ctx, opts) assert.NoError(t, err) diff --git a/cluster/calcium/create.go b/cluster/calcium/create.go index 2e44f1c39..3562e83f7 100644 --- a/cluster/calcium/create.go +++ b/cluster/calcium/create.go @@ -330,11 +330,9 @@ func (c *Calcium) doDeployOneWorkload(ctx context.Context, node *types.Node, opt } logger.Infof(ctx, "workload %s metadata created", workload.ID) - if len(opts.Files) > 0 { - for _, file := range opts.Files { - if err = c.doSendFileToWorkload(ctx, node.Engine, workload.ID, file); err != nil { - return err - } + for _, file := range opts.Files { + if err = c.doSendFileToWorkload(ctx, node.Engine, workload.ID, file); err != nil { + return err } } @@ -410,7 +408,6 @@ func (c *Calcium) doMakeWorkloadOptions(ctx context.Context, no int, msg *types. createOpts := &enginetypes.VirtualizationCreateOptions{} createOpts.EngineParams = msg.EngineParams createOpts.RawArgs = opts.RawArgs - createOpts.Lambda = opts.Lambda createOpts.User = opts.User createOpts.DNS = opts.DNS createOpts.Image = opts.Image diff --git a/cluster/calcium/create_test.go b/cluster/calcium/create_test.go index b3355812e..9a75a32df 100644 --- a/cluster/calcium/create_test.go +++ b/cluster/calcium/create_test.go @@ -14,7 +14,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" plugintypes "github.com/projecteru2/core/resource/plugins/types" resourcetypes "github.com/projecteru2/core/resource/types" @@ -336,10 +335,7 @@ func TestDoDeployWorkloadsOnNodeErrorPerWorkload(t *testing.T) { store := c.store.(*storemocks.Store) store.On("GetNode", mock.Anything, mock.Anything).Return(node, nil) store.On("ListNodeWorkloads", mock.Anything, mock.Anything, mock.Anything).Return(nil, types.ErrMockError) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) engine.On("VirtualizationCreate", mock.Anything, mock.Anything).Return(nil, types.ErrMockError) engine.On("VirtualizationInspect", mock.Anything, mock.Anything).Return(nil, types.ErrWorkloadNotExists) @@ -495,10 +491,7 @@ func newCreateWorkloadCluster(t *testing.T, createProcessingErr, deleteProcessin store.On("CreateProcessing", mock.Anything, mock.Anything, mock.Anything).Return(createProcessingErr) store.On("DeleteProcessing", mock.Anything, mock.Anything, mock.Anything).Return(deleteProcessingErr) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNodesByPod", mock.Anything, mock.Anything, mock.Anything).Return(nodes, nil) store.On("GetNode", mock.Anything, mock.Anything).Return( diff --git a/cluster/calcium/dissociate_test.go b/cluster/calcium/dissociate_test.go index 96819b0a6..e14a52ca2 100644 --- a/cluster/calcium/dissociate_test.go +++ b/cluster/calcium/dissociate_test.go @@ -1,14 +1,12 @@ package calcium import ( - "context" "testing" "testing/synctest" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" storemocks "github.com/projecteru2/core/store/mocks" @@ -23,9 +21,7 @@ func TestDissociateWorkload(t *testing.T) { store := c.store.(*storemocks.Store) rmgr := c.rmgr.(*resourcemocks.Manager) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) c1 := &types.Workload{ Resources: resourcetypes.Resources{}, diff --git a/cluster/calcium/lambda.go b/cluster/calcium/lambda.go index d66226be6..038d3252d 100644 --- a/cluster/calcium/lambda.go +++ b/cluster/calcium/lambda.go @@ -21,7 +21,6 @@ func (c *Calcium) RunAndWait(ctx context.Context, opts *types.DeployOptions, inC workloadIDs := []string{} logger := log.WithFunc("calcium.RunAndWait").WithField("app", opts.Name) - opts.Lambda = true if opts.OpenStdin && (opts.Count != 1 || opts.DeployStrategy != strategy.Auto) { logger.Errorf(ctx, types.ErrRunAndWaitCountOneWithStdin, "count %d method %s", opts.Count, opts.DeployStrategy) return workloadIDs, nil, types.ErrRunAndWaitCountOneWithStdin @@ -144,7 +143,7 @@ func (c *Calcium) RunAndWait(ctx context.Context, opts *types.DeployOptions, inC func newEruErrMsg(workloadID, format string, args ...any) *types.AttachWorkloadMessage { return &types.AttachWorkloadMessage{ WorkloadID: workloadID, - Data: []byte(fmt.Sprintf(format, args...)), + Data: fmt.Appendf(nil, format, args...), StdStreamType: types.EruError, } } diff --git a/cluster/calcium/lambda_test.go b/cluster/calcium/lambda_test.go index fc64a6f74..221ee7f4c 100644 --- a/cluster/calcium/lambda_test.go +++ b/cluster/calcium/lambda_test.go @@ -13,7 +13,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" plugintypes "github.com/projecteru2/core/resource/plugins/types" resourcetypes "github.com/projecteru2/core/resource/types" @@ -289,10 +288,7 @@ func newLambdaCluster(t *testing.T) (*Calcium, []*types.Node) { store.On("UpdateProcessing", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) store.On("DeleteProcessing", mock.Anything, mock.Anything, mock.Anything).Return(nil) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNodesByPod", mock.Anything, mock.Anything, mock.Anything).Return(nodes, nil) store.On("GetNode", mock.Anything, diff --git a/cluster/calcium/lock_test.go b/cluster/calcium/lock_test.go index ac04ca800..14d5102d4 100644 --- a/cluster/calcium/lock_test.go +++ b/cluster/calcium/lock_test.go @@ -237,9 +237,7 @@ func TestWithNodesPlanLockedTakesPodAndNodeLocksInKeyOrder(t *testing.T) { {NodeMeta: types.NodeMeta{Name: "c1", Podname: "podb"}, Available: true}, } store.On("GetNodes", mock.Anything, mock.Anything).Return(nodes, nil) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) keys := []string{} store.On("CreateLock", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { keys = append(keys, args.String(0)) @@ -257,9 +255,7 @@ func TestWithNodesPlanLockedTakesOnlyTheNodeLockForOneCandidate(t *testing.T) { c := NewTestCluster() store := c.store.(*storemocks.Store) store.On("GetNode", mock.Anything, "a1").Return(&types.Node{NodeMeta: types.NodeMeta{Name: "a1", Podname: "poda"}, Available: true}, nil) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) keys := []string{} store.On("CreateLock", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { keys = append(keys, args.String(0)) @@ -339,10 +335,7 @@ func TestWithNodeOperationLocked(t *testing.T) { }, Available: true, } - lock := &lockmocks.DistributedLock{} - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) - lock.On("Unlock", mock.Anything).Return(nil) - lock.On("Lock", mock.Anything).Return(context.Background(), nil) + mockLocks(t, store) store.On("GetNode", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() err := c.withNodeOperationLocked(ctx, "test", func(ctx context.Context, node *types.Node) error { return nil }) assert.Error(t, err) diff --git a/cluster/calcium/network.go b/cluster/calcium/network.go index 671445b42..0c91e352d 100644 --- a/cluster/calcium/network.go +++ b/cluster/calcium/network.go @@ -30,7 +30,7 @@ func (c *Calcium) ListNetworks(ctx context.Context, podname, driver string) ([]* drivers = append(drivers, driver) } - // every node of a pod reports the same networks + // Every node of a pod reports the same networks node := nodes[0] networks, err = node.Engine.NetworkList(ctx, drivers) diff --git a/cluster/calcium/node.go b/cluster/calcium/node.go index 71386f64d..72579ceb3 100644 --- a/cluster/calcium/node.go +++ b/cluster/calcium/node.go @@ -106,7 +106,7 @@ func (c *Calcium) RemoveNode(ctx context.Context, nodename string) error { _, txnErr := utils.Txn(ctx, func(ctx context.Context) error { - // a down node has no status key, so peers miss the removal unless one is written first + // A down node has no status key, so peers miss the removal unless one is written first if err = c.store.SetNodeStatus(ctx, node, 90); err != nil { logger.Warnf(ctx, "failed to set node status: %s", err) } @@ -239,7 +239,7 @@ func (c *Calcium) SetNode(ctx context.Context, opts *types.SetNodeOptions) (*typ if updateErr := c.store.UpdateNodes(ctx, node); updateErr != nil { return updateErr } - // capacity refresh is best effort; the store write already succeeded + // Capacity refresh is best effort; the store write already succeeded if len(opts.Resources) != 0 { _ = c.refreshResourceInfo(ctx, node) } diff --git a/cluster/calcium/node_test.go b/cluster/calcium/node_test.go index 13b8f34bb..d8977794c 100644 --- a/cluster/calcium/node_test.go +++ b/cluster/calcium/node_test.go @@ -13,7 +13,6 @@ import ( "github.com/projecteru2/core/engine/factory" enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" storemocks "github.com/projecteru2/core/store/mocks" @@ -129,10 +128,7 @@ func TestRemoveNode(t *testing.T) { ctx := t.Context() store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) name := "test" node := &types.Node{NodeMeta: types.NodeMeta{Name: name}} store.On("GetNode", mock.Anything, mock.Anything).Return(node, nil) @@ -290,10 +286,7 @@ func TestSetNode(t *testing.T) { assert.Error(t, err) store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) name := "test" opts.Nodename = name node := &types.Node{NodeMeta: types.NodeMeta{Name: name}} diff --git a/cluster/calcium/pod_test.go b/cluster/calcium/pod_test.go index a8f44ebf8..862ffa6ff 100644 --- a/cluster/calcium/pod_test.go +++ b/cluster/calcium/pod_test.go @@ -1,13 +1,11 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -39,10 +37,7 @@ func TestRemovePod(t *testing.T) { assert.Error(t, c.RemovePod(ctx, "")) store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("RemovePod", mock.Anything, mock.Anything).Return(nil) store.On("GetNodesByPod", mock.Anything, mock.Anything, mock.Anything).Return( []*types.Node{{NodeMeta: types.NodeMeta{Name: "test"}}}, nil, diff --git a/cluster/calcium/raw_engine_test.go b/cluster/calcium/raw_engine_test.go index 1d59ee10c..d59d1178e 100644 --- a/cluster/calcium/raw_engine_test.go +++ b/cluster/calcium/raw_engine_test.go @@ -1,7 +1,6 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" @@ -9,7 +8,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -18,10 +16,7 @@ func TestRawEngine(t *testing.T) { c := NewTestCluster() ctx := t.Context() store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) workload := &types.Workload{ ID: "id1", Privileged: true, diff --git a/cluster/calcium/realloc_test.go b/cluster/calcium/realloc_test.go index 937010789..bde9eda3a 100644 --- a/cluster/calcium/realloc_test.go +++ b/cluster/calcium/realloc_test.go @@ -10,7 +10,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" storemocks "github.com/projecteru2/core/store/mocks" @@ -27,9 +26,7 @@ func TestRealloc(t *testing.T) { rmgr.On("GetNodeResourceInfo", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, nil, nil) c.config.Scheduler.ShareBase = 100 - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) engine := &enginemocks.API{} engine.On("VirtualizationInspect", mock.Anything, mock.Anything).Return(&enginetypes.VirtualizationInfo{}, nil) @@ -118,9 +115,7 @@ func TestReallocJournalsRepairEntries(t *testing.T) { }) c.wal = mwal - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) engine := &enginemocks.API{} engine.On("VirtualizationUpdateResource", mock.Anything, mock.Anything, mock.Anything).Return(nil) node := &types.Node{NodeMeta: types.NodeMeta{Name: "node1"}, Engine: engine} @@ -161,9 +156,7 @@ func TestReallocRepairsRuntimeBeforeCommittingItsJournal(t *testing.T) { node := &types.Node{NodeMeta: types.NodeMeta{Name: "node1"}, Engine: engine} workload := &types.Workload{ID: "c1", Nodename: node.Name, Engine: engine, EngineParams: oldParams} - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) store := c.store.(*storemocks.Store) store.On("UpdateWorkload", mock.Anything, mock.Anything).Return(nil) store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) @@ -218,10 +211,7 @@ func TestReallocKeepsRepairEntriesUntilRollbackCompletes(t *testing.T) { workload := &types.Workload{ID: "c1", Nodename: node.Name, Engine: engine, Resources: resourcetypes.Resources{}} store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("UpdateWorkload", mock.Anything, workload).Return(types.ErrMockError).Once() store.On("UpdateWorkload", mock.Anything, mock.Anything).Return(nil).Once() rmgr := c.rmgr.(*resourcemocks.Manager) diff --git a/cluster/calcium/remap.go b/cluster/calcium/remap.go index d2e4e50fc..9d93059a9 100644 --- a/cluster/calcium/remap.go +++ b/cluster/calcium/remap.go @@ -25,7 +25,7 @@ func (c *Calcium) RemapResourceAndLog(ctx context.Context, logger *log.Fields, n } } -// the caller must hold the node lock +// The caller must hold the node lock func (c *Calcium) doRemapResource(ctx context.Context, logger *log.Fields, node *types.Node) error { engineParamsMap, workloads, err := c.computeRemap(ctx, node) if err != nil || len(engineParamsMap) == 0 { diff --git a/cluster/calcium/remap_test.go b/cluster/calcium/remap_test.go index 595fc57a1..6f2d6047e 100644 --- a/cluster/calcium/remap_test.go +++ b/cluster/calcium/remap_test.go @@ -1,7 +1,6 @@ package calcium import ( - "context" "strconv" "sync/atomic" "testing" @@ -11,7 +10,6 @@ import ( "github.com/stretchr/testify/require" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" "github.com/projecteru2/core/log" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" @@ -45,10 +43,7 @@ func TestRemapResource(t *testing.T) { assert.Nil(t, c.doRemapResource(t.Context(), log.WithField("test", "zc"), node)) store.On("GetNode", mock.Anything, mock.Anything).Return(node, nil) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) c.RemapResourceAndLog(t.Context(), log.WithField("test", "zc"), node.Name) } @@ -131,10 +126,7 @@ func TestRemapReplayRecomputesFromLiveState(t *testing.T) { workload := &types.Workload{ID: "workload1", Nodename: node.Name, Engine: engine} store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNode", mock.Anything, node.Name).Return(node, nil) store.On("NotFound", mock.Anything).Return(false) store.On("ListNodeWorkloads", mock.Anything, node.Name, mock.Anything).Return([]*types.Workload{workload}, nil).Twice() diff --git a/cluster/calcium/remove_test.go b/cluster/calcium/remove_test.go index 76566bbca..e1fe3aebf 100644 --- a/cluster/calcium/remove_test.go +++ b/cluster/calcium/remove_test.go @@ -25,9 +25,7 @@ func TestRemoveWorkload(t *testing.T) { c := NewTestCluster() defer c.pool.Release() ctx := t.Context() - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) store := c.store.(*storemocks.Store) rmgr := c.rmgr.(*resourcemocks.Manager) rmgr.On("GetNodeResourceInfo", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, nil, nil) @@ -143,9 +141,7 @@ func TestRemoveWorkloadJournalsRepairEntries(t *testing.T) { }) c.wal = mwal - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) engine := &enginemocks.API{} engine.On("VirtualizationRemove", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) workload := &types.Workload{ID: "xx", Name: "test", Nodename: "test", Engine: engine} @@ -182,9 +178,7 @@ func TestRemoveWorkloadKeepsTheNodeEntryWhenTheReleaseFails(t *testing.T) { }) c.wal = mwal - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) engine := &enginemocks.API{} engine.On("VirtualizationRemove", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) workload := &types.Workload{ID: "xx", Name: "test", Nodename: "test", Engine: engine} @@ -221,9 +215,7 @@ func TestRemoveWorkloadKeepsTheNodeEntryWhenTheRemovalFails(t *testing.T) { }) c.wal = mwal - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) engine := &enginemocks.API{} engine.On("VirtualizationRemove", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.ErrMockError) workload := &types.Workload{ID: "xx", Name: "test", Nodename: "test", Engine: engine} @@ -267,9 +259,7 @@ func TestRemoveWorkloadLocksTheWorkloadThenItsNode(t *testing.T) { store.On("ListNodeWorkloads", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) rmgr.On("SetNodeResourceUsage", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(resourcetypes.Resources{}, resourcetypes.Resources{}, nil) rmgr.On("Remap", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) var mu sync.Mutex keys := []string{} store.On("CreateLock", mock.Anything, mock.Anything).Run(func(args mock.Arguments) { diff --git a/cluster/calcium/replace_test.go b/cluster/calcium/replace_test.go index c4b36dc99..471c17ced 100644 --- a/cluster/calcium/replace_test.go +++ b/cluster/calcium/replace_test.go @@ -1,7 +1,6 @@ package calcium import ( - "context" "sync/atomic" "testing" @@ -10,7 +9,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" @@ -21,9 +19,7 @@ import ( func TestReplaceWorkload(t *testing.T) { c := NewTestCluster() ctx := t.Context() - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + lock := heldLock(t) store := c.store.(*storemocks.Store) rmgr := c.rmgr.(*resourcemocks.Manager) rmgr.On("GetNodeResourceInfo", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, nil, nil, nil) diff --git a/cluster/calcium/resource.go b/cluster/calcium/resource.go index 517bceb54..c08cf74fe 100644 --- a/cluster/calcium/resource.go +++ b/cluster/calcium/resource.go @@ -129,10 +129,5 @@ func (c *Calcium) doGetDeployStrategy(ctx context.Context, nodenames []string, o }) } - deployMap, err := strategy.Deploy(ctx, opts.DeployStrategy, opts.Count, opts.NodesLimit, strategyInfos, total) - if err != nil { - return nil, err - } - - return deployMap, nil + return strategy.Deploy(ctx, opts.DeployStrategy, opts.Count, opts.NodesLimit, strategyInfos, total) } diff --git a/cluster/calcium/resource_test.go b/cluster/calcium/resource_test.go index 768be9105..f89378089 100644 --- a/cluster/calcium/resource_test.go +++ b/cluster/calcium/resource_test.go @@ -1,14 +1,12 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" storemocks "github.com/projecteru2/core/store/mocks" @@ -75,10 +73,7 @@ func TestNodeResource(t *testing.T) { nodename := "testnode" store := c.store.(*storemocks.Store) rmgr := c.rmgr.(*resourcemocks.Manager) - lock := &lockmocks.DistributedLock{} - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) + mockLocks(t, store) node := &types.Node{ NodeMeta: types.NodeMeta{ diff --git a/cluster/calcium/send_test.go b/cluster/calcium/send_test.go index 1f6e848c3..7a82bf06a 100644 --- a/cluster/calcium/send_test.go +++ b/cluster/calcium/send_test.go @@ -1,14 +1,12 @@ package calcium import ( - "context" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -32,10 +30,7 @@ func TestSend(t *testing.T) { }, } store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetWorkloads", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() store.On("GetWorkload", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() ch, err := c.Send(ctx, opts) diff --git a/cluster/calcium/sendlarge_test.go b/cluster/calcium/sendlarge_test.go index d6b8662d7..befe06813 100644 --- a/cluster/calcium/sendlarge_test.go +++ b/cluster/calcium/sendlarge_test.go @@ -9,7 +9,6 @@ import ( "github.com/stretchr/testify/mock" enginemocks "github.com/projecteru2/core/engine/mocks" - lockmocks "github.com/projecteru2/core/lock/mocks" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" ) @@ -27,10 +26,7 @@ func TestSendLarge(t *testing.T) { optsChan := make(chan *types.SendLargeFileOptions) store := &storemocks.Store{} c.store = store - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetWorkloads", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() store.On("GetWorkload", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() ch := c.SendLargeFile(ctx, optsChan) diff --git a/cluster/calcium/status.go b/cluster/calcium/status.go index 7a6b353e6..4c19c608e 100644 --- a/cluster/calcium/status.go +++ b/cluster/calcium/status.go @@ -41,7 +41,7 @@ func (c *Calcium) GetWorkloadsStatus(ctx context.Context, IDs []string) ([]*type func (c *Calcium) SetWorkloadsStatus(ctx context.Context, statusMetas []*types.StatusMeta, ttls map[string]int64) ([]*types.StatusMeta, error) { logger := log.WithFunc("calcium.SetWorkloadsStatus").WithField("count", len(statusMetas)).WithField("ttls", ttls) - // old callers omit appname, nodename and entrypoint; look them up + // Old callers omit appname, nodename and entrypoint; look them up missing := []string{} for _, statusMeta := range statusMetas { if statusMeta.Appname == "" || statusMeta.Nodename == "" || statusMeta.Entrypoint == "" { diff --git a/cluster/calcium/wal_test.go b/cluster/calcium/wal_test.go index 553fd4873..2a10d4632 100644 --- a/cluster/calcium/wal_test.go +++ b/cluster/calcium/wal_test.go @@ -18,7 +18,6 @@ import ( enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" - lockmocks "github.com/projecteru2/core/lock/mocks" "github.com/projecteru2/core/log" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" @@ -37,10 +36,7 @@ func TestHandleWorkloadResourceAllocatedMultipleNodes(t *testing.T) { c := NewTestCluster() store := c.store.(*storemocks.Store) rmgr := c.rmgr.(*resourcemocks.Manager) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("ListNodeWorkloads", mock.Anything, mock.Anything, mock.Anything).Return(nil, types.ErrMockError) store.On("GetNode", mock.Anything, mock.Anything).Return( func(_ context.Context, name string) *types.Node { @@ -65,10 +61,7 @@ func TestHandleWorkloadResourceAllocatedKeepsEntryUntilEveryNodeIsFixed(t *testi c := NewTestCluster() enableTestWAL(t, c) store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNode", mock.Anything, "n1").Return(&types.Node{NodeMeta: types.NodeMeta{Name: "n1"}}, nil) store.On("ListNodeWorkloads", mock.Anything, "n1", mock.Anything).Return(nil, types.ErrMockError).Twice() @@ -292,10 +285,7 @@ func TestHandleReallocWorkload(t *testing.T) { engine := &enginemocks.API{} engineParams := resourcetypes.Resources{"cpumem": {"cpu": 2}} store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNode", mock.Anything, "n1").Return(&types.Node{NodeMeta: types.NodeMeta{Name: "n1"}}, nil) store.On("GetWorkload", mock.Anything, "workloadid").Return( &types.Workload{ID: "workloadid", Nodename: "n1", EngineParams: engineParams, Engine: engine}, nil, @@ -319,10 +309,7 @@ func TestHandleReallocWorkloadOnAnEngineThatCannotReplayIt(t *testing.T) { engine := &enginemocks.API{} engineParams := resourcetypes.Resources{"cpumem": {"cpu": 2}} store := c.store.(*storemocks.Store) - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) store.On("GetNode", mock.Anything, "n1").Return(&types.Node{NodeMeta: types.NodeMeta{Name: "n1"}}, nil) store.On("GetWorkload", mock.Anything, "workloadid").Return( &types.Workload{ID: "workloadid", Nodename: "n1", EngineParams: engineParams, Engine: engine}, nil, @@ -390,10 +377,7 @@ func TestHandleCreateLambda(t *testing.T) { Return(nil). Once() store.On("ListNodeWorkloads", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil).Once() - lock := &lockmocks.DistributedLock{} - lock.On("Lock", mock.Anything).Return(context.Background(), nil) - lock.On("Unlock", mock.Anything).Return(nil) - store.On("CreateLock", mock.Anything, mock.Anything).Return(lock, nil) + mockLocks(t, store) c.wal.Recover(t.Context()) synctest.Wait() diff --git a/engine/containerd/create.go b/engine/containerd/create.go index ae3211a39..4ba1992d2 100644 --- a/engine/containerd/create.go +++ b/engine/containerd/create.go @@ -92,7 +92,7 @@ func (e *Engine) VirtualizationCreate(ctx context.Context, opts *enginetypes.Vir } } - // the container id is the workload name: containerd carries no name eru-agent could read + // The container id is the workload name: containerd carries no name eru-agent could read ID := opts.Name if err := identifiers.Validate(ID); err != nil { return nil, errors.Wrapf(coretypes.ErrInvalidWorkloadName, "containerd cannot name %q", ID) diff --git a/engine/containerd/dockerfile.go b/engine/containerd/dockerfile.go index 95c0442c1..26325a48f 100644 --- a/engine/containerd/dockerfile.go +++ b/engine/containerd/dockerfile.go @@ -124,7 +124,7 @@ func preparedSource(ctx context.Context, build *enginetypes.Build, scm coresourc } } - // artifacts replace the cloned tree so no source ships in the image + // Artifacts replace the cloned tree so no source ships in the image if len(build.Artifacts) > 0 { artifactsDir := buildDir if cloneDir != "" { diff --git a/engine/containerd/lifecycle.go b/engine/containerd/lifecycle.go index 897b91604..090dc6158 100644 --- a/engine/containerd/lifecycle.go +++ b/engine/containerd/lifecycle.go @@ -38,7 +38,7 @@ func (e *Engine) VirtualizationStart(ctx context.Context, ID string) (err error) if err != nil { return err } - // a task has fifos or a log uri, never both: an interactive workload takes the node fifos + // A task has fifos or a log uri, never both: an interactive workload takes the node fifos creator := cio.LogURI(logShimURL) var relay *attach if _, stdin := info.Labels[stdinLabel]; stdin { @@ -231,7 +231,7 @@ func (e *Engine) VirtualizationUpdateResource(ctx context.Context, ID string, en return err } limits := resourceSpec(resource, &RawArgs{}, devices) - // live first, stored spec second: a failure then persists nothing, and a restart replays the new limits only after both held + // Live first, stored spec second: a failure then persists nothing, and a restart replays the new limits only after both held task, err := optionalTask(ctx, found) if err != nil { return err diff --git a/engine/containerd/logs.go b/engine/containerd/logs.go index 43dd15f18..1e9972aae 100644 --- a/engine/containerd/logs.go +++ b/engine/containerd/logs.go @@ -139,9 +139,9 @@ func (e *Engine) relayFifos(ctx context.Context, ID string) (_ cio.Creator, _ *a return nil, nil, err } - // a relay lives as long as the workload, and the deploy request that starts it does not + // A relay lives as long as the workload, and the deploy request that starts it does not held := context.WithoutCancel(ctx) - // the shim holds the stdin fifo open read-write, so only CloseIO ends the workload's input + // The shim holds the stdin fifo open read-write, so only CloseIO ends the workload's input relay := &attach{ died: make(chan error, relayStreams), closeStdin: func(ctx context.Context) error { return e.closeTaskStdin(ctx, ID) }, diff --git a/engine/containerd/spec.go b/engine/containerd/spec.go index 17d6a0206..c049222cf 100644 --- a/engine/containerd/spec.go +++ b/engine/containerd/spec.go @@ -414,7 +414,7 @@ func passwdEntry(passwd, user string) ([]string, bool) { } func additionalGids(group, user string, primary uint32) []uint32 { - gids := []uint32{} + var gids []uint32 for line := range strings.Lines(group) { fields := strings.Split(strings.TrimRight(line, "\n"), ":") if len(fields) <= groupMembers || !slices.Contains(strings.Split(fields[groupMembers], ","), user) { @@ -426,9 +426,6 @@ func additionalGids(group, user string, primary uint32) []uint32 { } gids = append(gids, uint32(gid)) } - if len(gids) == 0 { - return nil - } return gids } @@ -496,12 +493,8 @@ func hostNetworking(networks map[string]string) bool { } func withoutNetSysctls(sysctl map[string]string) map[string]string { - kept := make(map[string]string, len(sysctl)) - for key, value := range sysctl { - if !strings.HasPrefix(key, netSysctlPrefix) { - kept[key] = value - } - } + kept := maps.Clone(sysctl) + maps.DeleteFunc(kept, func(key, _ string) bool { return strings.HasPrefix(key, netSysctlPrefix) }) return kept } diff --git a/engine/factory/factory.go b/engine/factory/factory.go index de32d2122..dbdd6bcac 100644 --- a/engine/factory/factory.go +++ b/engine/factory/factory.go @@ -144,7 +144,7 @@ func (e *EngineCache) checkNodeStatus(ctx context.Context) { } if !ns.Alive { - // one node may back several engines + // One node may back several engines e.cache.Range(func(_, v any) bool { ep := v.(engine.API).GetParams() if ep.Nodename == ns.Nodename { @@ -232,26 +232,22 @@ func validateEngine(ctx context.Context, engine engine.API, timeout time.Duratio return err } -func getEnginePrefix(endpoint string) (string, error) { - for prefix := range engines { +func engineFactory(endpoint string) (factory, error) { + for prefix, newClient := range engines { if strings.HasPrefix(endpoint, prefix) { - return prefix, nil + return newClient, nil } } - return "", errors.Wrapf(types.ErrInvaildNodeEndpoint, "endpoint invalid %+v", endpoint) + return nil, errors.Wrapf(types.ErrInvaildNodeEndpoint, "endpoint invalid %+v", endpoint) } func newEngine(ctx context.Context, config types.Config, params *enginetypes.Params) (client engine.API, err error) { - prefix, err := getEnginePrefix(params.Endpoint) + newClient, err := engineFactory(params.Endpoint) if err != nil { return nil, err } - e, ok := engines[prefix] - if !ok { - return nil, types.ErrInvaildEngineEndpoint - } utils.WithTimeout(ctx, config.ConnectionTimeout, func(ctx context.Context) { - client, err = e(ctx, config, params.Nodename, params.Endpoint) + client, err = newClient(ctx, config, params.Nodename, params.Endpoint) }) if err != nil { return nil, err diff --git a/engine/process/unit.go b/engine/process/unit.go index 1fa8df867..29d1a23c7 100644 --- a/engine/process/unit.go +++ b/engine/process/unit.go @@ -181,7 +181,7 @@ func cpuWeight(quota float64, remap bool) int { return max(1, int(math.Round(defaultCPUWeight*fraction))) } -// a bind needs no RootDirectory, so raw workloads carry them too +// A bind needs no RootDirectory, so raw workloads carry them too func bindPaths(binds []utils.VolumeBind) []string { props := make([]string, 0, len(binds)) for _, bind := range binds { diff --git a/engine/sshrunner/ssh.go b/engine/sshrunner/ssh.go index 68e452724..b37d2fa14 100644 --- a/engine/sshrunner/ssh.go +++ b/engine/sshrunner/ssh.go @@ -141,12 +141,12 @@ func (r *sshRunner) Files(ctx context.Context) (Files, error) { } func (r *sshRunner) Dial(ctx context.Context, network, addr string) (net.Conn, error) { - // a forward is not a session, so MaxSessions does not bound it + // A forward is not a session, so MaxSessions does not bound it return retry(ctx, r, func(client *ssh.Client) (net.Conn, error) { return client.Dial(network, addr) }) } func (r *sshRunner) Ping(ctx context.Context) error { - // a global request on the connection answers even when every session is held + // A global request on the connection answers even when every session is held _, err := retry(ctx, r, func(client *ssh.Client) (struct{}, error) { _, _, err := client.SendRequest(keepaliveRequest, true, nil) return struct{}{}, err diff --git a/engine/types/virtualization.go b/engine/types/virtualization.go index 0f49cc87e..2f1543560 100644 --- a/engine/types/virtualization.go +++ b/engine/types/virtualization.go @@ -28,7 +28,6 @@ type VirtualizationCreateOptions struct { Networks map[string]string RawArgs []byte - Lambda bool } // VirtualizationCreated identifies a freshly created workload. diff --git a/go.mod b/go.mod index 5acc375f3..afdae12b4 100644 --- a/go.mod +++ b/go.mod @@ -36,7 +36,6 @@ require ( go.etcd.io/etcd/client/pkg/v3 v3.7.1 go.etcd.io/etcd/client/v3 v3.7.1 go.etcd.io/etcd/server/v3 v3.7.1 - go.uber.org/zap v1.28.0 golang.org/x/crypto v0.56.0 golang.org/x/sync v0.22.0 google.golang.org/grpc v1.83.2 @@ -145,6 +144,7 @@ require ( go.opentelemetry.io/proto/otlp v1.11.0 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect + go.uber.org/zap v1.28.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/exp v0.0.0-20260820142414-ca536658362e // indirect diff --git a/log/log.go b/log/log.go index 96df9b84a..03b66ee70 100644 --- a/log/log.go +++ b/log/log.go @@ -54,7 +54,7 @@ func Fatalf(ctx context.Context, err error, format string, args ...any) { func logWriter(cfg *types.ServerLogConfig) io.Writer { switch { case cfg.Filename != "": - // file log always uses json format + // File log always uses json format return &lumberjack.Logger{ Filename: cfg.Filename, MaxBackups: cfg.MaxBackups, diff --git a/metrics/metrics.go b/metrics/metrics.go index 1abf3274a..63c9b9c98 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -51,7 +51,7 @@ func (m *Metrics) SendDeployCount(ctx context.Context, n int) { metrics := &plugintypes.Metrics{ Name: deployCountName, Labels: []string{m.Hostname}, - Key: deployCountKey, + Key: fmt.Sprintf(deployCountKey, m.Hostname), Value: strconv.Itoa(n), } diff --git a/metrics/metrics_test.go b/metrics/metrics_test.go index d85404d0e..b92ca19d2 100644 --- a/metrics/metrics_test.go +++ b/metrics/metrics_test.go @@ -1,11 +1,14 @@ package metrics import ( + "net" "sync" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" plugintypes "github.com/projecteru2/core/resource/plugins/types" ) @@ -44,6 +47,28 @@ func TestSendMetricsConcurrentStatsd(t *testing.T) { assert.Equal(t, 1, collected(gauge)) } +func TestSendDeployCountNamesTheHost(t *testing.T) { + conn, err := net.ListenPacket("udp", "127.0.0.1:0") + require.NoError(t, err) + defer func() { _ = conn.Close() }() + m := &Metrics{ + Hostname: "host1", + StatsdAddr: conn.LocalAddr().String(), + Collectors: map[string]prometheus.Collector{ + deployCountName: prometheus.NewCounterVec(prometheus.CounterOpts{Name: deployCountName}, []string{"hostname"}), + }, + } + + m.SendDeployCount(t.Context(), 3) + require.NoError(t, m.statsdClient.Close()) + + buf := make([]byte, 512) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) + n, _, err := conn.ReadFrom(buf) + require.NoError(t, err) + assert.Equal(t, "core.host1.deploy.count:3|c", string(buf[:n])) +} + func collected(c prometheus.Collector) int { ch := make(chan prometheus.Metric, 16) c.Collect(ch) diff --git a/resource/cobalt/node.go b/resource/cobalt/node.go index d4fa4b804..50cff2df4 100644 --- a/resource/cobalt/node.go +++ b/resource/cobalt/node.go @@ -27,7 +27,7 @@ func (m *Manager) AddNode(ctx context.Context, nodename string, opts resourcetyp func(ctx context.Context) error { resps, err := call(ctx, m.plugins, func(plugin plugins.Plugin) (*plugintypes.AddNodeResponse, error) { r := opts[plugin.Name()] - // plugins run even for a nil request: they read config from engine info and seed an empty etcd entry + // Plugins run even for a nil request: they read config from engine info and seed an empty etcd entry logger.WithField("plugin", plugin.Name()).Debugf(ctx, "add node request %+v", r) return plugin.AddNode(ctx, nodename, r, nodeInfo) }) @@ -262,9 +262,6 @@ func (m *Manager) getNodeResourceInfo(ctx context.Context, nodename string, ps [ resourceDiffs := []string{} resps, err := call(ctx, ps, func(plugin plugins.Plugin) (*plugintypes.GetNodeResourceInfoResponse, error) { - var resp *plugintypes.GetNodeResourceInfoResponse - var err error - name := plugin.Name() wrks := make([]plugintypes.WorkloadResource, 0, len(workloads)) for _, wrk := range workloads { @@ -272,11 +269,9 @@ func (m *Manager) getNodeResourceInfo(ctx context.Context, nodename string, ps [ } if fix { - resp, err = plugin.FixNodeResource(ctx, nodename, wrks) - } else { - resp, err = plugin.GetNodeResourceInfo(ctx, nodename, wrks) + return plugin.FixNodeResource(ctx, nodename, wrks) } - return resp, err + return plugin.GetNodeResourceInfo(ctx, nodename, wrks) }) if err != nil { return nil, nil, nil, err diff --git a/resource/plugins/cpumem/calculate.go b/resource/plugins/cpumem/calculate.go index 9608dea43..318636dfe 100644 --- a/resource/plugins/cpumem/calculate.go +++ b/resource/plugins/cpumem/calculate.go @@ -70,7 +70,7 @@ func (p Plugin) CalculateRealloc(ctx context.Context, nodename string, resource return nil, err } - // the origin usage returns to the pool before the new total is computed + // The origin usage returns to the pool before the new total is computed nodeResourceInfo.Usage.Sub(&cpumemtypes.NodeResource{ CPU: originResource.CPURequest, CPUMap: originResource.CPUMap, diff --git a/resource/plugins/cpumem/calculate_test.go b/resource/plugins/cpumem/calculate_test.go index 429af5e7f..2eb93ae7a 100644 --- a/resource/plugins/cpumem/calculate_test.go +++ b/resource/plugins/cpumem/calculate_test.go @@ -4,7 +4,6 @@ import ( "fmt" "testing" - "github.com/cockroachdb/errors" "github.com/docker/go-units" "github.com/stretchr/testify/assert" @@ -178,7 +177,7 @@ func TestCalculateRemap(t *testing.T) { } _, err = cm.CalculateRemap(ctx, "xxx", workloadsResource) - assert.True(t, errors.Is(err, coretypes.ErrKeyNotFound)) + assert.ErrorIs(t, err, coretypes.ErrKeyNotFound) r, err := cm.CalculateRemap(ctx, node, workloadsResource) assert.Nil(t, err) diff --git a/resource/plugins/cpumem/cpumem_test.go b/resource/plugins/cpumem/cpumem_test.go index 41c2dd6a7..60e0df88f 100644 --- a/resource/plugins/cpumem/cpumem_test.go +++ b/resource/plugins/cpumem/cpumem_test.go @@ -38,11 +38,6 @@ func generateNodes(ctx context.Context, t testing.TB, cm *Plugin, nums, cores in assert.NoError(t, err) names = append(names, name) } - t.Cleanup(func() { - for name := range reqs { - cm.RemoveNode(ctx, name) - } - }) return names } diff --git a/resource/plugins/cpumem/node.go b/resource/plugins/cpumem/node.go index 7120ec7de..4edd314e4 100644 --- a/resource/plugins/cpumem/node.go +++ b/resource/plugins/cpumem/node.go @@ -368,10 +368,7 @@ func (p Plugin) getKnown(ctx context.Context, keys []string) (map[string]string, } func (p Plugin) doGetNodesResourceInfo(ctx context.Context, nodenames []string) (map[string]*cpumemtypes.NodeResourceInfo, error) { - keys := make([]string, 0, len(nodenames)) - for _, nodename := range nodenames { - keys = append(keys, fmt.Sprintf(nodeResourceInfoKey, nodename)) - } + keys := utils.Map(nodenames, func(nodename string) string { return fmt.Sprintf(nodeResourceInfoKey, nodename) }) data, err := p.store.GetMulti(ctx, keys) if err != nil { return nil, err @@ -433,7 +430,7 @@ func (p Plugin) calculateNodeResource(req *cpumemtypes.NodeResourceRequest, node var resp *cpumemtypes.NodeResource if origin == nil || !delta { // no delta means node resource rewrite with whole new data resp = &cpumemtypes.NodeResource{CPUMap: cpumemtypes.CPUMap{}, NUMAMemory: cpumemtypes.NUMAMemory{}, NUMA: cpumemtypes.NUMA{}} - // a full rewrite must add onto the zero value; subtracting would store negative amounts + // A full rewrite must add onto the zero value; subtracting would store negative amounts incr = true } else { resp = origin.DeepCopy() diff --git a/resource/plugins/cpumem/node_test.go b/resource/plugins/cpumem/node_test.go index 2b58f5947..84e5a316a 100644 --- a/resource/plugins/cpumem/node_test.go +++ b/resource/plugins/cpumem/node_test.go @@ -6,7 +6,6 @@ import ( "math" "testing" - "github.com/cockroachdb/errors" "github.com/docker/go-units" "github.com/stretchr/testify/assert" @@ -93,7 +92,7 @@ func TestGetNodesDeployCapacityWithCPUBind(t *testing.T) { } _, err := cm.GetNodesDeployCapacity(ctx, []string{"xxx"}, req) - assert.True(t, errors.Is(err, coretypes.ErrKeyNotFound)) + assert.ErrorIs(t, err, coretypes.ErrKeyNotFound) tests := []struct { name string @@ -202,7 +201,7 @@ func TestGetNodesDeployCapacityWithMemory(t *testing.T) { } _, err := cm.GetNodesDeployCapacity(ctx, nodes, req) - assert.True(t, errors.Is(err, types.ErrInvalidMemory)) + assert.ErrorIs(t, err, types.ErrInvalidMemory) tests := []struct { name string @@ -351,7 +350,7 @@ func TestGetAndFixNodeResourceInfo(t *testing.T) { node := nodes[0] _, err := cm.GetNodeResourceInfo(ctx, "xxx", nil) - assert.True(t, errors.Is(err, coretypes.ErrKeyNotFound)) + assert.ErrorIs(t, err, coretypes.ErrKeyNotFound) r, err := cm.GetNodeResourceInfo(ctx, node, nil) assert.Nil(t, err) @@ -590,7 +589,6 @@ func BenchmarkGetNodesCapacityScaling(b *testing.B) { cm := initCPUMEM(b) nodes := generateNodes(ctx, b, cm, tc.nodes, tc.cores, 128*units.GB, 100, 0) req := plugintypes.WorkloadResourceRequest{"cpu-bind": true, "cpu-request": 1.3, "memory-request": "1"} - b.ResetTimer() for b.Loop() { if _, err := cm.GetNodesDeployCapacity(ctx, nodes, req); err != nil { b.Fatal(err) diff --git a/resource/plugins/cpumem/schedule/schedule.go b/resource/plugins/cpumem/schedule/schedule.go index 686242edc..99e8cad12 100644 --- a/resource/plugins/cpumem/schedule/schedule.go +++ b/resource/plugins/cpumem/schedule/schedule.go @@ -66,7 +66,7 @@ func newHost(cpuMap types.CPUMap, shareBase, maxFragmentCores int) *host { } } - // busier cores go first so idle cores stay whole + // Busier cores go first so idle cores stay whole slices.SortStableFunc(h.fullCores, byLoad) slices.SortStableFunc(h.fragmentCores, byLoad) @@ -143,10 +143,7 @@ func (h *host) bestSplit(full, fragment, maxMoved int) (int, int) { for moved := 1; moved <= maxMoved; moved++ { fragmentCapacity[moved] = fragmentCapacity[moved-1] + h.fullCores[moved-1].pieces/fragment } - fullCapacities := make([]int, maxMoved+1) - for i := range fullCapacities { - fullCapacities[i] = -1 - } + fullCapacities := slices.Repeat([]int{-1}, maxMoved+1) fullCapacity := func(moved int) int { if fullCapacities[moved] < 0 { fullCapacities[moved] = h.countFullCPUPlans(h.fullCores[moved:], full) @@ -190,7 +187,7 @@ func (h *host) getFullCPUPlans(cores []*cpuCore, full int) []types.CPUMap { plans = append(plans, ranked{plan, rank}) }) if !h.affinity { - // restore the pre-heap core priority across the produced plans + // Restore the pre-heap core priority across the produced plans slices.SortFunc(plans, func(a, b ranked) int { return cmp.Compare(a.rank, b.rank) }) } return utils.Map(plans, func(r ranked) types.CPUMap { return r.plan }) diff --git a/resource/plugins/cpumem/types/node.go b/resource/plugins/cpumem/types/node.go index 96f035666..e2c697f05 100644 --- a/resource/plugins/cpumem/types/node.go +++ b/resource/plugins/cpumem/types/node.go @@ -1,6 +1,7 @@ package types import ( + "cmp" "maps" "strconv" "strings" @@ -125,7 +126,7 @@ func (n *NodeResourceInfo) Validate() error { } } - // the stored record always carries objects, never nulls + // The stored record always carries objects, never nulls for _, r := range []*NodeResource{n.Capacity, n.Usage} { if r.CPUMap == nil { r.CPUMap = CPUMap{} @@ -164,10 +165,7 @@ func (n *NodeResourceRequest) Parse(config coretypes.Config, rawParams resourcet } if cpu := rawParams.Int64("cpu"); cpu > 0 { - share := rawParams.Int64("share") - if share == 0 { - share = int64(config.Scheduler.ShareBase) - } + share := cmp.Or(rawParams.Int64("share"), int64(config.Scheduler.ShareBase)) for i := range cpu { n.CPUMap[strconv.FormatInt(i, 10)] = int(share) diff --git a/resource/plugins/cpumem/types/workload.go b/resource/plugins/cpumem/types/workload.go index b8bb7b7c3..72fa3752c 100644 --- a/resource/plugins/cpumem/types/workload.go +++ b/resource/plugins/cpumem/types/workload.go @@ -97,7 +97,7 @@ func (w *WorkloadResourceRequest) Validate() error { if w.CPURequest > 0 && w.CPULimit > 0 && w.CPULimit < w.CPURequest { w.CPULimit = w.CPURequest } - // a cpu-bound workload gets request raised to limit, never the other way round + // A cpu-bound workload gets request raised to limit, never the other way round if w.CPUBind && w.CPURequest > 0 && w.CPULimit > 0 && w.CPULimit > w.CPURequest { w.CPURequest = w.CPULimit } diff --git a/rpc/rpc.go b/rpc/rpc.go index ab4da692a..cacfa1678 100644 --- a/rpc/rpc.go +++ b/rpc/rpc.go @@ -3,6 +3,7 @@ package rpc import ( "archive/tar" "bufio" + "cmp" "context" "fmt" "io" @@ -458,32 +459,30 @@ func (v *Vibranium) Copy(opts *pb.CopyOptions, stream pb.CoreRPC_CopyServer) err } r, w := io.Pipe() - utils.SentryGo(func(m *types.CopyMessage) func() { - return func() { - var err error - defer func() { - w.CloseWithError(err) //nolint:errcheck - }() - - tw := tar.NewWriter(w) - defer func() { err = errors.Join(err, tw.Close()) }() - header := &tar.Header{ - Name: filepath.Base(m.Filename), - Uid: m.UID, - Gid: m.GID, - Mode: m.Mode, - Size: int64(len(m.Content)), - } - if err = tw.WriteHeader(header); err != nil { - logger.Error(task.context, err, "write tarball header") - return - } - if _, err = tw.Write(m.Content); err != nil { - logger.Error(task.context, err, "write tarball content") - return - } + utils.SentryGo(func() { + var err error + defer func() { + w.CloseWithError(err) //nolint:errcheck + }() + + tw := tar.NewWriter(w) + defer func() { err = errors.Join(err, tw.Close()) }() + header := &tar.Header{ + Name: filepath.Base(m.Filename), + Uid: m.UID, + Gid: m.GID, + Mode: m.Mode, + Size: int64(len(m.Content)), + } + if err = tw.WriteHeader(header); err != nil { + logger.Error(task.context, err, "write tarball header") + return } - }(m)) + if _, err = tw.Write(m.Content); err != nil { + logger.Error(task.context, err, "write tarball content") + return + } + }) for { n, err := r.Read(p) @@ -800,23 +799,20 @@ func (v *Vibranium) LogStream(opts *pb.LogStreamOptions, stream pb.CoreRPC_LogSt func (v *Vibranium) RunAndWait(stream pb.CoreRPC_RunAndWaitServer) error { task := v.newTask(stream.Context(), "RunAndWait", true) - RunAndWaitOptions, deployOpts, err := runAndWaitOptions(stream) + runOpts, deployOpts, err := runAndWaitOptions(stream) if err != nil { task.done() return grpcstatus.Error(RunAndWait, err.Error()) } logger := log.WithFunc("vibranium.RunAndWait") - opts := RunAndWaitOptions.DeployOptions + opts := runOpts.DeployOptions var ( ctx context.Context cancel context.CancelFunc ) - if RunAndWaitOptions.Async { - timeout := v.config.GlobalTimeout - if RunAndWaitOptions.AsyncTimeout != 0 { - timeout = time.Second * time.Duration(RunAndWaitOptions.AsyncTimeout) - } + if runOpts.Async { + timeout := cmp.Or(time.Duration(runOpts.AsyncTimeout)*time.Second, v.config.GlobalTimeout) ctx, cancel = context.WithTimeout(context.WithoutCancel(task.context), timeout) // task.done cancels task.context } else { ctx, cancel = context.WithCancel(task.context) @@ -846,7 +842,7 @@ func (v *Vibranium) RunAndWait(stream pb.CoreRPC_RunAndWaitServer) error { f(ch) } - if !RunAndWaitOptions.Async { + if !runOpts.Async { runAndWait(func(ch <-chan *types.AttachWorkloadMessage) { for m := range ch { if err = stream.Send(toRPCAttachWorkloadMessage(m)); err != nil { @@ -970,19 +966,19 @@ func drainUntilStop[T, R any](t *task, name string, code codes.Code, stop <-chan } func runAndWaitOptions(stream pb.CoreRPC_RunAndWaitServer) (*pb.RunAndWaitOptions, *types.DeployOptions, error) { - RunAndWaitOptions, err := stream.Recv() + runOpts, err := stream.Recv() if err != nil { return nil, nil, err } - if RunAndWaitOptions.DeployOptions == nil { + if runOpts.DeployOptions == nil { return nil, nil, types.ErrNoDeployOpts } - opts := RunAndWaitOptions.DeployOptions - if RunAndWaitOptions.Async { + opts := runOpts.DeployOptions + if runOpts.Async { opts.OpenStdin = false } deployOpts, err := toCoreDeployOptions(opts) - return RunAndWaitOptions, deployOpts, err + return runOpts, deployOpts, err } func reallocResult(err error) (*pb.ReallocResourceMessage, error) { diff --git a/rpc/rpc_test.go b/rpc/rpc_test.go index 415036cef..b240a0f84 100644 --- a/rpc/rpc_test.go +++ b/rpc/rpc_test.go @@ -35,11 +35,7 @@ func TestAddNode(t *testing.T) { v := newVibranium() opts := &pb.AddNodeOptions{} cluster := v.cluster.(*clustermock.Cluster) - cluster.On("AddNode", mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - ).Return(nil, types.ErrMockError).Once() + cluster.On("AddNode", mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() _, err := v.AddNode(t.Context(), opts) assert.Error(t, err) engine := &enginemock.API{} @@ -50,16 +46,12 @@ func TestAddNode(t *testing.T) { Engine: engine, } engine.On("Info", mock.Anything).Return(&enginetypes.Info{}, nil) - cluster.On("AddNode", mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - mock.Anything, mock.Anything, mock.Anything, mock.Anything, - ).Return(node, nil) + cluster.On("AddNode", mock.Anything, mock.Anything).Return(node, nil) _, err = v.AddNode(t.Context(), opts) assert.NoError(t, err) } -func TestSetNodeTranform(t *testing.T) { +func TestSetNodeTransform(t *testing.T) { b := &pb.SetNodeOptions{ Nodename: "a", } @@ -68,7 +60,7 @@ func TestSetNodeTranform(t *testing.T) { assert.Equal(t, "a", opts.Nodename) } -func TestSetNodeTranformRejectsAMalformedResource(t *testing.T) { +func TestSetNodeTransformRejectsAMalformedResource(t *testing.T) { b := &pb.SetNodeOptions{ Nodename: "a", Resources: map[string][]byte{"cpumem": []byte("not json")}, @@ -209,9 +201,9 @@ func TestGetWorkloadReportsAMissingWorkloadAsNotFound(t *testing.T) { cluster.On("GetWorkload", mock.Anything, "broken"). Return(nil, types.ErrMockError).Once() - _, err := v.GetWorkload(context.Background(), &pb.WorkloadID{Id: "gone"}) + _, err := v.GetWorkload(t.Context(), &pb.WorkloadID{Id: "gone"}) assert.Equal(t, codes.NotFound, grpcstatus.Code(err)) - _, err = v.GetWorkload(context.Background(), &pb.WorkloadID{Id: "broken"}) + _, err = v.GetWorkload(t.Context(), &pb.WorkloadID{Id: "broken"}) assert.Equal(t, GetWorkload, grpcstatus.Code(err)) } diff --git a/rpc/transform.go b/rpc/transform.go index 0ca5e0b8a..0a75bec57 100644 --- a/rpc/transform.go +++ b/rpc/transform.go @@ -297,7 +297,6 @@ func toCoreDeployOptions(d *pb.DeployOptions) (*types.DeployOptions, error) { ExtraHosts: d.ExtraHosts, Networks: d.Networks, User: d.User, - Debug: d.Debug, OpenStdin: d.OpenStdin, Labels: d.Labels, DeployStrategy: d.DeployStrategy.String(), diff --git a/scripts/check_calico.py b/scripts/check_calico.py deleted file mode 100755 index 576d01140..000000000 --- a/scripts/check_calico.py +++ /dev/null @@ -1,171 +0,0 @@ -#!/usr/bin/env python3 -import argparse -import json -import os -import re -import subprocess -import sys - -import etcd3 - -def sh(prog, *args): - p = subprocess.Popen((prog,)+args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True) - so, se = p.communicate() - return p.returncode, so, se - - -class Calico(object): - - @classmethod - def get_weps(cls): - items = cls.get('wep', '--all-namespaces') - return WorkloadEndpoint.parse(items) - - @classmethod - def get(cls, resource_type, *args): - rc, so, se = cls.ctl('get', resource_type, '-o', 'json', *args) - if rc: - raise ValueError('get %s failed: %s; %s; %s' % (resource_type, rc, so, se)) - return json.loads(so.decode('utf-8'))['items'] - - @classmethod - def ctl(cls, subcommand, *args): - return sh('calicoctl', subcommand, *args) - - -class WorkloadEndpoint(object): - - def __init__(self, raw_dict): - self.dict = raw_dict - - @classmethod - def parse(cls, items): - weps = [] - for elem in items: - weps.append(WorkloadEndpoint(elem)) - return weps - - def has_belong_eru(self, eru): - return any(eru.has_ip(ip) for ip in self.ips) - - @property - def name(self): - return self.dict['metadata']['name'] - - @property - def namespace(self): - return self.dict['metadata']['namespace'] - - @property - def node(self): - return self.dict['spec']['node'] - - @property - def interface(self): - return self.dict['spec']['interfaceName'] - - @property - def ips(self): - return [cidr.split('/')[0] for cidr in self.dict['spec']['ipNetworks']] - - -class Eru(object): - - def __init__(self, etcd, root_prefix): - self.etcd = etcd - self.root_prefix = os.path.join('/', root_prefix.lstrip('/')) - self.ips = [] - - def watch_ips(self): - - ipre = re.compile(r'^(\d+\.\d+\.\d+\.\d+)') - - def parse(key, value): - for _, ip in json.loads(value).get('networks', {}).items(): - match = ipre.search(ip) - if not match: - continue - self.ips.append(match.group(1)) - - self.etcd.get_prefix(self.workload_status_prefix, parse) - - @property - def workload_status_prefix(self): - # the ended with '/' is necessary to avoid 'status:node' - return os.path.join(self.root_prefix, 'status/') - - def has_ip(self, ip): - return ip in self.ips - - -class ETCD(object): - - def __init__(self, cli): - self.cli = cli - - @classmethod - def connect(cls, host, port): - cli = etcd3.client(host=host, port=port) - return ETCD(cli) - - def get_prefix(self, prefix, fn): - start = prefix - end = etcd3.utils.increment_last_byte(etcd3.utils.to_bytes(start)) - - while 1: - req = etcd3.etcdrpc.RangeRequest() - req.key = etcd3.utils.to_bytes(start) - req.keys_only = False - req.range_end = etcd3.utils.to_bytes(end) - req.sort_order = etcd3.etcdrpc.RangeRequest.ASCEND - req.sort_target = etcd3.etcdrpc.RangeRequest.KEY - req.serializable = True - req.limit = 1000 - - resp = self.cli.kvstub.Range( - req, - self.cli.timeout, - credentials=self.cli.call_credentials, - metadata=self.cli.metadata, - ) - - for kv in resp.kvs: - key = kv.key.decode('utf-8') - fn(kv.key.decode('utf-8'), kv.value.decode('utf-8')) - - if not resp.more: - return - - start = etcd3.utils.increment_last_byte(kv.key) - - -def print_dangling(wep): - print('%s/%s is dangling' % (wep.namespace, wep.name)) - -def get_args(): - ap = argparse.ArgumentParser() - ap.add_argument('-e', '--eru-etcd-endpoints', help='the ERU ETCD endpoints', default='127.0.0.1:2379') - ap.add_argument('-p', '--eru-etcd-prefix', help='the ERU ETCD root prefix', required=True) - return ap.parse_args() - -def main(): - args = get_args() - - host, _, port = args.eru_etcd_endpoints.split(',')[0].partition(':') - port = int(port) if port else 2379 - etcd = ETCD.connect(host, port) - - global eru - eru = Eru(etcd, args.eru_etcd_prefix) - eru.watch_ips() - - for wep in Calico.get_weps(): - if 'yavirt-cali-gw' in wep.interface or wep.has_belong_eru(eru): - continue - - print_dangling(wep) - - return 0 - -if __name__ == '__main__': - sys.exit(main()) diff --git a/scripts/meta_transfer_as_rename2workload.py b/scripts/meta_transfer_as_rename2workload.py deleted file mode 100644 index 1d7f1d104..000000000 --- a/scripts/meta_transfer_as_rename2workload.py +++ /dev/null @@ -1,379 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -import argparse -import functools -import json -import os - -import etcd3 -import sys - -check_conflict = False -dry_run = False -override = False - - -def remove_prefix(s, prefix): - return s[len(prefix):].lstrip('/') if s.startswith(prefix) else s - - -def range_prefix(meta, obj_prefix, fn): - etcd = meta.etcd - orig_prefix = os.path.join(meta.orig_root_prefix, obj_prefix) - range_start = orig_prefix - range_end = etcd3.utils.increment_last_byte( - etcd3.utils.to_bytes(range_start) - ) - - while True: - range_request = etcd3.etcdrpc.RangeRequest() - range_request.key = etcd3.utils.to_bytes(range_start) - range_request.keys_only = False - range_request.range_end = etcd3.utils.to_bytes(range_end) - range_request.sort_order = etcd3.etcdrpc.RangeRequest.ASCEND - range_request.sort_target = etcd3.etcdrpc.RangeRequest.KEY - range_request.serializable = True - range_request.limit = 1000 - - range_response = etcd.kvstub.Range( - range_request, - etcd.timeout, - credentials=etcd.call_credentials, - metadata=etcd.metadata, - ) - - for kv in range_response.kvs: - orig_key = kv.key.decode('utf-8') - objname = remove_prefix(orig_key, orig_prefix) - new_key = fn(objname, kv.value.decode('utf-8')) - if not new_key: - continue - - if check_conflict: - if etcd.get(new_key)[0]: - print('conflict: new key exists! original: %s, new: %s' % (orig_key, new_key)) - else: - print('convert %s to %s' % (orig_key, new_key)) - - if not range_response.more: - break - - range_start = etcd3.utils.increment_last_byte(kv.key) - - -def exists(client, key): - if override: - return False - return client.get(key)[0] is not None - - -class Pod(object): - - def __init__(self, meta, podname=None): - """Initializes a pod transfer.""" - self.meta = meta - self.pod_prefix = 'pod/info' - self.range_prefix = functools.partial(range_prefix, self.meta) - self.podname = podname - self.exists = functools.partial(exists, self.meta.etcd) - - def trans(self): - self.range_prefix(self.pod_prefix, self._trans) - - def _trans(self, podname, orig_value): - # Only trans the specific pod if any. - if self.podname and self.podname != podname: - return - - new_key = os.path.join(self.meta.new_root_prefix, self.pod_prefix, podname) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, orig_value) - return new_key - - -class Node(object): - - def __init__(self, meta, podname): - """Initializes a node transfer.""" - self.meta = meta - self.info_prefix = 'node' - self.range_prefix = functools.partial(range_prefix, self.meta, self.info_prefix) - self.pod_nodes = {} - self.podname = podname - self.exists = functools.partial(exists, self.meta.etcd) - - def trans(self): - self.range_prefix(self._trans_pod) - self.range_prefix(self._trans_info) - self.range_prefix(self._trans_cert) - self.range_prefix(self._trans_workload) - return self.pod_nodes.keys() - - def _trans_info(self, nodename, orig_value): - # skipping extra info. - if ':' in nodename: - return - - if not self._belong_pod(nodename): - return - - self.pod_nodes[nodename] = json.loads(orig_value) - - new_key = os.path.join(self.meta.new_root_prefix, self.info_prefix, nodename) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, orig_value) - return new_key - - def _trans_pod(self, node_pod_pair, orig_value): - # parsering node-pod-pair itself only - if ':pod/' not in node_pod_pair: - return - - podname, _, nodename = node_pod_pair.partition(':pod/') - if not (podname and nodename): - raise ValueError('invalid podname or nodename for %s' % node_pod_pair) - - # Only trans the specific pod if any. - if self.podname and self.podname != podname: - return - - self.pod_nodes[nodename] = {} - - new_key = os.path.join(self.meta.new_root_prefix, self.info_prefix, '%s:pod' % podname, nodename) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, orig_value) - return new_key - - def _trans_cert(self, cert_key, orig_value): - nodename, _, cert_type = cert_key.partition(':') - if not self._belong_pod(nodename): - return - - # parsering orig_key which ends with :ca, :cert, :key only - if cert_type not in ('ca', 'cert', 'key'): - return - - new_key = os.path.join(self.meta.new_root_prefix, self.info_prefix, '%s:%s' % (nodename, cert_type)) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, orig_value) - return new_key - - def _trans_workload(self, node_wrk_pair, orig_value): - nodename, _, wrk_id = node_wrk_pair.partition(':containers/') - # parsering orig_key which belongs node-workload pair only. - if not (nodename and wrk_id): - return - - if not self._belong_pod(nodename): - return - - new_key = os.path.join(self.meta.new_root_prefix, self.info_prefix, '%s:workloads' % nodename, wrk_id) - wrk = Workload.conv(orig_value, self) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, json.dumps(wrk)) - return new_key - - def _belong_pod(self, nodename): - if not self.podname: - return True - return nodename in self.pod_nodes.keys() - - def get_numa_node(self, cpumap, nodename): - """Ref core types/core.go GetNUMANode func.""" - numa_node_id = "" - - node = self.pod_nodes.get(nodename) - if not node: - raise ValueError('invalid nodename %s' % nodename) - - numa = node.get('numa') - if not numa: - return numa_node_id - - for cpu_id in cpumap: - mem_node = numa.get(cpu_id) - if not mem_node: - continue - - if numa_node_id == '': - numa_node_id = mem_node - elif numa_node_id != mem_node: - numa_node_id = '' - - return numa_node_id - - -class Workload(object): - - def __init__(self, meta, node_transfer, podname=None, pod_nodes=None): - """Initializes a workload transfer.""" - self.meta = meta - self.container_prefix = 'containers' - self.wrk_prefix = 'workloads' - self.deploy_prefix = 'deploy' - self.range_prefix = functools.partial(range_prefix, self.meta) - self.node_transfer = node_transfer - self.podname = podname - self.pod_nodes = pod_nodes - self.workloads = set() - self.exists = functools.partial(exists, self.meta.etcd) - - def trans(self): - self.range_prefix(self.deploy_prefix, self._trans_deploy) - self.range_prefix(self.container_prefix, self._trans_container) - - def _trans_container(self, wrk_id, orig_value): - # skipping if the workload doesn't belong the specific pod. - if not self.podname or wrk_id not in self.workloads: - return - - new_key = os.path.join(self.meta.new_root_prefix, self.wrk_prefix, wrk_id) - wrk = self.conv(orig_value, self.node_transfer) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, json.dumps(wrk)) - return new_key - - def _trans_deploy(self, deploy_key, orig_value): - parts = deploy_key.split('/') - if len(parts) != 4: - print('invalid deploy key: %s' % deploy_key) - return - - appname, entrypoint, nodename, wrk_id = parts - if not self._belong_pod(nodename): - return - - self.workloads.update({wrk_id}) - - new_key = os.path.join(self.meta.new_root_prefix, self.deploy_prefix, appname, entrypoint, nodename, wrk_id) - wrk = self.conv(orig_value, self.node_transfer) - if not dry_run and not self.exists(new_key): - self.meta.etcd.put(new_key, json.dumps(wrk)) - - return new_key - - def _belong_pod(self, nodename): - if not self.podname: - return True - return nodename in self.pod_nodes - - @classmethod - def conv(cls, orig_value, node_transfer): - - def delete(*keys): - for k in keys: - try: - del dic[k] - except KeyError: - pass - - del_keys = set() - - def get(new_field, orig_field, transit_field, default=None): - value = None - - # don't use dic.get(new_field, dic[orig_field]), - # due to there isn't orig_field but has new_field. - if new_field in dic: - value = dic[new_field] - elif transit_field in dic: - value = dic[transit_field] - else: - if default is None: - value = dic[orig_field] - else: - value = dic.get(orig_field, default) - - del_keys.update({orig_field, transit_field}) - - return value - - dic = json.loads(orig_value) - dic['cpu'] = get('CPU', 'cpu', 'CPU') - - dic.update(dict( - create_time=1553990400, - cpu_quota_request=get('cpu_quota_request', 'quota', 'CPUQuotaRequest'), - cpu_quota_limit=get('cpu_quota_limit', 'quota', 'CPUQuotaLimit'), - memory_request=get('memory_request', 'memory', 'MemoryRequest'), - memory_limit=get('memory_limit', 'memory', 'MemoryLimit'), - volume_request=get('volume_request', 'volumes', 'VolumeRequest', default=[]), - volume_limit=get('volume_limit', 'volumes', 'VolumeLimit', default=[]), - volume_plan_request=get('volume_plan_request', 'volume_plan', 'VolumePlanRequest', default={}), - volume_plan_limit=get('volume_plan_limit', 'volume_plan', 'VolumePlanLimit', default={}), - volume_changed=dic.get('volume_changed', False), - storage_request=get('storage_request', 'storage', 'StorageRequest'), - storage_limit=get('storage_limit', 'storage', 'StorageLimit'), - )) - - dic['numa_node'] = '' - if dic['cpu'] and node_transfer: - numa_node = node_transfer.get_numa_node(dic['cpu'], dic['nodename']) - dic['numa_node'] = dic.get('NUMANode', numa_node) - - # don't removing *cpu* from the original dict. - try: - del_keys.remove('cpu') - except KeyError: - pass - - del_keys.update({'softlimit', 'VolumeChanged', 'NUMANode'}) - delete(*list(del_keys)) - - return dic - - -class Transfer(object): - - def __init__(self, etcd, orig_root_prefix, new_root_prefix): - """Initializes a transfer which includes common utilities.""" - self.etcd = etcd - self.orig_root_prefix = orig_root_prefix - self.new_root_prefix = new_root_prefix - - def trans(self, podname=None): - Pod(self, podname).trans() - - node_transfer = Node(self, podname) - nodes = node_transfer.trans() - - Workload(self, node_transfer, podname, nodes).trans() - - -def getargs(): - ap = argparse.ArgumentParser() - ap.add_argument('-o', '--orig', dest='orig_root_prefix', help='original prefix', default='/eru') - ap.add_argument('-n', '--new', dest='new_root_prefix', help='new prefix', default='/eru2') - ap.add_argument('-p', '--pod', dest='podname') - ap.add_argument('--etcd-host', default='127.0.0.1') - ap.add_argument('--etcd-port', type=int, default=2379) - ap.add_argument('--dry-run', dest='dry_run', action='store_true', help='dry run, will not actually migrate') - ap.add_argument('--check-conflict', dest='check_conflict', action='store_true', - help='check conflict, checks if destination already has the key') - ap.add_argument('--override', dest='override', action='store_true', - help='if set, the value will be migrated anyway no matter if it already exists') - return ap.parse_args() - - -def connect_etcd(host, port): - return etcd3.client(host=host, port=port) - - -def main(): - args = getargs() - - global dry_run, check_conflict, override - dry_run = args.dry_run - check_conflict = args.check_conflict - if check_conflict: - dry_run = True - override = args.override - - etcd = connect_etcd(args.etcd_host, args.etcd_port) - trans = Transfer(etcd, args.orig_root_prefix, args.new_root_prefix) - trans.trans(args.podname) - return 0 - - -if __name__ == '__main__': - sys.exit(main()) diff --git a/scripts/meta_transfer_resource_plugin.py b/scripts/meta_transfer_resource_plugin.py deleted file mode 100644 index d468ecec0..000000000 --- a/scripts/meta_transfer_resource_plugin.py +++ /dev/null @@ -1,369 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -import argparse -import functools -import json -import os - -import etcd3 - -dry_run = False -record_prefix = "upgrade_" -origin_data_record_path = 'origin_data_record.data' -transferred_node_record_path = 'transferred_node_record.data' -transferred_workload_record_path = 'transferred_workload_record.data' - -origin_data_recorder = None -transferred_node_recorder = None -transferred_workload_recorder = None - -transferred_workloads = set() -transferred_nodes = set() - - -def init_recorders(): - global origin_data_recorder, origin_data_record_path - global transferred_node_recorder, transferred_node_record_path, transferred_nodes - global transferred_workload_recorder, transferred_workload_record_path, transferred_workloads - - origin_data_record_path = record_prefix + origin_data_record_path - transferred_node_record_path = record_prefix + transferred_node_record_path - transferred_workload_record_path = record_prefix + transferred_workload_record_path - - if os.path.exists(transferred_node_record_path): - with open(transferred_node_record_path, 'r') as f: - transferred_nodes = set(f.read().strip('\n').splitlines()) - - if os.path.exists(transferred_workload_record_path): - with open(transferred_workload_record_path, 'r') as f: - transferred_workloads = set(f.read().strip('\n').splitlines()) - - origin_data_recorder = open(origin_data_record_path, 'a') - transferred_node_recorder = open(transferred_node_record_path, 'a') - transferred_workload_recorder = open(transferred_workload_record_path, 'a') - - -def close_recorders(): - transferred_node_recorder.close() - transferred_workload_recorder.close() - origin_data_recorder.close() - - -def add_record(recorder, record): - recorder.write('%s\n' % record) - - -def remove_prefix(s, prefix): - return s[len(prefix):].lstrip('/') if s.startswith(prefix) else s - - -def dict_sub(d1, d2): - if d1 is None: - return None - if d2 is None: - return d1 - get = lambda d, k: d[k] if k in d else 0 - return {k: d1[k] - get(d2, k) for k in d1} - - -class ETCD: - def __init__(self, client, prefix): - """Create an instance of ETCD.""" - self.etcd = client - self.prefix = prefix - - def get(self, key): - if not key.startswith(self.prefix): - key = self.prefix + key - res = self.etcd.get(key)[0] - if res is None: - return None - return res.decode('utf-8') - - def put(self, key, value): - if not key.startswith(self.prefix): - key = self.prefix + key - if dry_run: - print('put {}\n{}'.format(key, value)) - return - - origin_value = self.get(key) - if origin_value: - add_record(origin_data_recorder, key) - add_record(origin_data_recorder, origin_value) - - self.etcd.put(key, value) - - def range_prefix(self, obj_prefix, fn): - prefix = self.prefix + obj_prefix - range_start = prefix - range_end = etcd3.utils.increment_last_byte( - etcd3.utils.to_bytes(range_start) - ) - - while True: - range_request = etcd3.etcdrpc.RangeRequest() - range_request.key = etcd3.utils.to_bytes(range_start) - range_request.keys_only = False - range_request.range_end = etcd3.utils.to_bytes(range_end) - range_request.sort_order = etcd3.etcdrpc.RangeRequest.ASCEND - range_request.sort_target = etcd3.etcdrpc.RangeRequest.KEY - range_request.serializable = True - range_request.limit = 1000 - - range_response = self.etcd.kvstub.Range( - range_request, - self.etcd.timeout, - credentials=self.etcd.call_credentials, - metadata=self.etcd.metadata, - ) - - for kv in range_response.kvs: - orig_key = kv.key.decode('utf-8') - objname = remove_prefix(orig_key, prefix) - fn(objname, kv.value.decode('utf-8')) - - if not range_response.more: - break - - range_start = etcd3.utils.increment_last_byte(kv.key) - - -etcd: ETCD = None - - -class Node: - def __init__(self, name, pod_name, meta): - """Initializes a node transfer.""" - self.name = name - self.pod_name = pod_name - self.meta = json.loads(meta) - - def upgrade(self): - cpumem_meta = self._gen_cpumem_meta() - volume_meta = self._gen_volume_meta() - cpumem_key = '/resource/cpumem/%s' % self.name - volume_key = '/resource/volume/%s' % self.name - etcd.put(cpumem_key, cpumem_meta) - etcd.put(volume_key, volume_meta) - - def downgrade(self): - self._load_resources_meta() - keys = ['/node/%s' % self.name, '/node/%s:pod/%s' % (self.pod_name, self.name)] - for key in keys: - etcd.put(key, json.dumps(self.meta)) - - def _load_cpumem_meta(self, meta): - cpumem_meta = json.loads(meta) - self.meta['init_cpu'] = cpumem_meta['capacity']['cpu_map'] - self.meta['cpu'] = dict_sub(cpumem_meta['capacity']['cpu_map'], cpumem_meta['usage']['cpu_map']) - self.meta['init_memcap'] = cpumem_meta['capacity']['memory'] - self.meta['memcap'] = cpumem_meta['capacity']['memory'] - cpumem_meta['usage']['memory'] - self.meta['cpuused'] = cpumem_meta['usage']['cpu'] - self.meta['numa'] = cpumem_meta['capacity']['numa'] - self.meta['init_numa_memory'] = cpumem_meta['capacity']['numa_memory'] - self.meta['numa_memory'] = dict_sub(cpumem_meta['capacity']['numa_memory'], cpumem_meta['usage']['numa_memory']) - - def _load_resources_meta(self): - # load cpumem resources - cpumem_key = '/resource/cpumem/%s' % self.name - cpumem_meta = etcd.get(cpumem_key) - if not cpumem_meta: - print("%s not found" % cpumem_key) - self._load_cpumem_meta(cpumem_meta) - - # load volume resources - volume_key = '/resource/volume/%s' % self.name - volume_meta = etcd.get(volume_key) - if not volume_meta: - print("%s not found" % volume_key) - self._load_volume_meta(volume_meta) - - def _load_volume_meta(self, meta): - volume_meta = json.loads(meta) - self.meta['init_volume'] = volume_meta['capacity']['volumes'] - self.meta['volume'] = volume_meta['usage']['volumes'] - self.meta['init_storage_cap'] = volume_meta['capacity']['storage'] - self.meta['storage_cap'] = volume_meta['usage']['storage'] - self.meta['volumeused'] = sum(volume_meta['usage']['volumes'].values()) - - def _gen_cpumem_meta(self): - cpumem_meta = {"capacity": {}, "usage": {}} - cpumem_meta['capacity']['cpu_map'] = self.meta['init_cpu'] - cpumem_meta['usage']['cpu_map'] = dict_sub(self.meta['init_cpu'], self.meta['cpu']) - cpumem_meta['capacity']['memory'] = self.meta['init_memcap'] - cpumem_meta['usage']['memory'] = self.meta['init_memcap'] - self.meta['memcap'] - cpumem_meta['capacity']['cpu'] = len(self.meta['init_cpu']) - cpumem_meta['usage']['cpu'] = self.meta['cpuused'] - cpumem_meta['capacity']['numa'] = self.meta['numa'] - cpumem_meta['capacity']['numa_memory'] = self.meta['init_numa_memory'] - cpumem_meta['usage']['numa_memory'] = dict_sub(self.meta['init_numa_memory'], self.meta['numa_memory']) - return json.dumps(cpumem_meta) - - def _gen_volume_meta(self): - volume_meta = {"capacity": {}, "usage": {}} - volume_meta['capacity']['volumes'] = self.meta['init_volume'] - volume_meta['usage']['volumes'] = dict_sub(self.meta['init_volume'], self.meta['volume']) - volume_meta['capacity']['storage'] = self.meta['init_storage_cap'] - volume_meta['usage']['storage'] = self.meta['init_storage_cap'] - self.meta['storage_cap'] - return json.dumps(volume_meta) - - -class Workload: - def __init__(self, workload_id, app_name, entry_name, node_name, meta): - """Initializes a workload transfer.""" - self.workload_id = workload_id - self.app_name = app_name - self.entry_name = entry_name - self.node_name = node_name - self.meta = json.loads(meta) - self.keys = ['/workloads/%s' % self.workload_id, - '/deploy/%s/%s/%s/%s' % (self.app_name, self.entry_name, self.node_name, self.workload_id), - '/node/%s:workloads/%s' % (self.node_name, self.workload_id)] - - def save(self): - for key in self.keys: - etcd.put(key, json.dumps(self.meta)) - - def upgrade(self): - if self.workload_id in transferred_workloads: - return - self._gen_resource_meta() - self.save() - - def downgrade(self): - if self.workload_id in transferred_workloads: - return - self._load_resource_meta() - self.save() - - def _gen_resource_meta(self): - self.meta['resource_args'] = {} - self.meta['resource_args']['cpumem'] = { - 'cpu_request': self.meta['cpu_quota_request'], - 'cpu_limit': self.meta['cpu_quota_limit'], - 'cpu_map': self.meta['cpu'], - 'memory_request': self.meta['memory_request'], - 'memory_limit': self.meta['memory_limit'], - "numa_node": self.meta['numa_node'], - } - self.meta['resource_args']['volume'] = { - 'volumes_request': self.meta['volume_request'], - 'volumes_limit': self.meta['volume_limit'], - 'volume_plan_request': self.meta['volume_plan_request'], - 'volume_plan_limit': self.meta['volume_plan_limit'], - 'storage_request': self.meta['storage_request'], - 'storage_limit': self.meta['storage_limit'], - } - self.meta['engine_args'] = { - 'cpu': self.meta['cpu_quota_limit'], - 'memory': self.meta['memory_limit'], - 'numa_node': self.meta['numa_node'], - 'cpu_map': self.meta['cpu'], - 'storage': self.meta['storage_limit'], - 'volume': [], - } - for binding in self.meta['volume_limit']: - if not binding.startswith('AUTO'): - self.meta['engine_args']['volume'].append(binding) - - for binding in self.meta['volume_plan_limit']: - groups = binding.split(':') - if len(groups) < 3: - print("volume plan limit of %s is invalid: %s" % (self.workload_id, binding)) - - dst = groups[1] - flags = groups[2] - device = list(self.meta['volume_plan_limit'][binding].keys())[0] - size = self.meta['volume_plan_limit'][binding][device] - flags = flags.replace('m', '') - if 'o' in flags: - flags = flags.replace('o', '').replace('r', 'ro').replace('w', 'wo') - - self.meta['engine_args']['volume'].append('%s:%s:%s:%s' % (device, dst, flags, size)) - - def _load_resource_meta(self): - self.meta['cpu_quota_request'] = self.meta['resource_args']['cpumem']['cpu_request'] - self.meta['cpu_quota_limit'] = self.meta['resource_args']['cpumem']['cpu_limit'] - self.meta['cpu'] = self.meta['resource_args']['cpumem']['cpu_map'] - self.meta['memory_request'] = self.meta['resource_args']['cpumem']['memory_request'] - self.meta['memory_limit'] = self.meta['resource_args']['cpumem']['memory_limit'] - self.meta['numa_node'] = self.meta['resource_args']['cpumem']['numa_node'] - self.meta['volume_request'] = self.meta['resource_args']['volume']['volumes_request'] - self.meta['volume_limit'] = self.meta['resource_args']['volume']['volumes_limit'] - self.meta['volume_plan_request'] = self.meta['resource_args']['volume']['volume_plan_request'] - self.meta['volume_plan_limit'] = self.meta['resource_args']['volume']['volume_plan_limit'] - self.meta['storage_request'] = self.meta['resource_args']['volume']['storage_request'] - self.meta['storage_limit'] = self.meta['resource_args']['volume']['storage_limit'] - - -def connect_etcd(host, port): - return etcd3.client(host=host, port=port) - - -def transfer_node(key, value, upgrade=True): - if ':pod' not in key: - return - node_name = key.split('/')[-1] - pod_name = key.split(':')[0].strip('/') - if node_name in transferred_nodes: - return - - print('transferring node %s' % node_name) - node = Node(node_name, pod_name, value) - if upgrade: - node.upgrade() - else: - node.downgrade() - add_record(transferred_node_recorder, node_name) - - -def transfer_workload(key, value, upgrade=True): - app_name, entry_name, node_name, workload_id = key.strip('/').split('/') - if workload_id in transferred_workloads: - return - - print('transferring workload %s' % workload_id) - workload = Workload(workload_id, app_name, entry_name, node_name, value) - if upgrade: - workload.upgrade() - else: - workload.downgrade() - add_record(transferred_workload_recorder, workload_id) - - -def transfer(upgrade=True): - etcd.range_prefix('/node', functools.partial(transfer_node, upgrade=upgrade)) - etcd.range_prefix('/deploy', functools.partial(transfer_workload, upgrade=upgrade)) - - -def get_args(): - ap = argparse.ArgumentParser() - ap.add_argument('--upgrade', action='store_true', help='upgrade to new eru-core') - ap.add_argument('--downgrade', action='store_true', help='downgrade to old eru-core') - ap.add_argument('--etcd-prefix', help='etcd prefix', default='/eru') - ap.add_argument('--etcd-host', default='127.0.0.1') - ap.add_argument('--etcd-port', type=int, default=2379) - ap.add_argument('--dry-run', dest='dry_run', action='store_true', help='dry run, will not actually migrate') - return ap.parse_args() - - -def main(): - args = get_args() - if not args.upgrade and not args.downgrade: - print('please specify --upgrade or --downgrade') - - global etcd, dry_run, record_prefix - etcd = ETCD(connect_etcd(args.etcd_host, args.etcd_port), args.etcd_prefix) - dry_run = args.dry_run - upgrade = args.upgrade - if not upgrade: - record_prefix = 'downgrade' - - init_recorders() - transfer(upgrade) - - -if __name__ == '__main__': - main() diff --git a/scripts/redis_as_broker.py b/scripts/redis_as_broker.py index ceccdec6a..92d483b5b 100644 --- a/scripts/redis_as_broker.py +++ b/scripts/redis_as_broker.py @@ -36,7 +36,7 @@ def migrate(root, etcd_client, redis_client, dry_run): if not key.startswith(root + '/'): continue - key = key.replace(root, '') + key = key[len(root):] value = kv.value.decode('utf-8') if dry_run: diff --git a/selfmon/selfmon.go b/selfmon/selfmon.go index 1801d6280..70f526ed8 100644 --- a/selfmon/selfmon.go +++ b/selfmon/selfmon.go @@ -184,7 +184,7 @@ func (n *NodeStatusWatcher) dealNodeStatusMessage(ctx context.Context, message * logger.Errorf(ctx, message.Error, "deal with node status stream message failed %+v", message) return } - // the agent owns the transition back to alive + // The agent owns the transition back to alive if message.Alive { return } diff --git a/store/common/deploy.go b/store/common/deploy.go index 87bf75264..9945497c6 100644 --- a/store/common/deploy.go +++ b/store/common/deploy.go @@ -14,7 +14,7 @@ func (s *Store) GetDeployStatus(ctx context.Context, appname, entryname string) var processingCount map[string]int g, gctx := errgroup.WithContext(ctx) g.Go(func() (err error) { - // trailing slash keeps the prefix from matching a longer entrypoint + // Trailing slash keeps the prefix from matching a longer entrypoint keys, err = s.ListPrefix(gctx, filepath.Join(WorkloadDeployPrefix, appname, entryname)+"/") return err }) diff --git a/store/common/node.go b/store/common/node.go index 7a305ecae..5d8eeb694 100644 --- a/store/common/node.go +++ b/store/common/node.go @@ -83,10 +83,7 @@ func (s *Store) GetNode(ctx context.Context, nodename string) (*types.Node, erro } func (s *Store) GetNodes(ctx context.Context, nodenames []string) ([]*types.Node, error) { - keys := make([]string, 0, len(nodenames)) - for _, nodename := range nodenames { - keys = append(keys, fmt.Sprintf(NodeInfoKey, nodename)) - } + keys := utils.Map(nodenames, func(nodename string) string { return fmt.Sprintf(NodeInfoKey, nodename) }) kvs, err := s.GetMulti(ctx, keys) if err != nil { diff --git a/store/common/processing.go b/store/common/processing.go index 9b20e1d71..5e965f918 100644 --- a/store/common/processing.go +++ b/store/common/processing.go @@ -19,7 +19,7 @@ func (s *Store) DeleteProcessing(ctx context.Context, processing *types.Processi func (s *Store) doLoadProcessing(ctx context.Context, appname, entryname string) (map[string]int, error) { nodesCount := map[string]int{} - // trailing slash keeps the prefix from matching a longer entrypoint + // Trailing slash keeps the prefix from matching a longer entrypoint data, err := s.GetPrefix(ctx, filepath.Join(WorkloadProcessingPrefix, appname, entryname)+"/", 0) if err != nil { return nil, err diff --git a/store/common/workload.go b/store/common/workload.go index 187c91875..cad2b651e 100644 --- a/store/common/workload.go +++ b/store/common/workload.go @@ -72,7 +72,7 @@ func (s *Store) ListWorkloads(ctx context.Context, appname, entrypoint, nodename if entrypoint == "" { nodename = "" } - // trailing slash keeps the prefix from matching a longer nodename + // Trailing slash keeps the prefix from matching a longer nodename data, err := s.GetPrefix(ctx, filepath.Join(WorkloadDeployPrefix, appname, entrypoint, nodename)+"/", limit) if err != nil { return nil, err @@ -95,7 +95,7 @@ func (s *Store) WorkloadStatusStream(ctx context.Context, appname, entrypoint, n if entrypoint == "" { nodename = "" } - // trailing slash keeps the prefix from matching a longer nodename + // Trailing slash keeps the prefix from matching a longer nodename statusKey := filepath.Join(WorkloadStatusPrefix, appname, entrypoint, nodename) + "/" ch := make(chan *types.WorkloadStatus) logger := log.WithFunc("store.common.WorkloadStatusStream") @@ -120,10 +120,7 @@ func (s *Store) getWorkload(ctx context.Context, ID string, withEngine bool) (*t } func (s *Store) getWorkloads(ctx context.Context, IDs []string, withEngine bool) ([]*types.Workload, error) { - keys := make([]string, 0, len(IDs)) - for _, ID := range IDs { - keys = append(keys, fmt.Sprintf(WorkloadInfoKey, ID)) - } + keys := utils.Map(IDs, func(ID string) string { return fmt.Sprintf(WorkloadInfoKey, ID) }) data, err := s.GetMulti(ctx, keys) if err != nil { @@ -194,7 +191,7 @@ func (s *Store) bindWorkloadsAdditions(ctx context.Context, workloads []*types.W if err != nil { return nil, err } - // trailing slash keeps the prefix from matching a longer nodename + // Trailing slash keeps the prefix from matching a longer nodename prefix := filepath.Join(WorkloadStatusPrefix, appname, entrypoint, workload.Nodename) + "/" groups[prefix] = append(groups[prefix], workload) nodenames[workload.Nodename] = struct{}{} diff --git a/store/etcdv3/meta/ephemeral_test.go b/store/etcdv3/meta/ephemeral_test.go index a8f455df2..10919b125 100644 --- a/store/etcdv3/meta/ephemeral_test.go +++ b/store/etcdv3/meta/ephemeral_test.go @@ -8,7 +8,7 @@ import ( ) func TestEphemeralDeregister(t *testing.T) { - m := NewEmbeddedETCD(t) + m := newEmbeddedETCD(t) ctx := t.Context() path := "/ident" @@ -34,7 +34,7 @@ func TestEphemeralDeregister(t *testing.T) { } func TestEphemeral(t *testing.T) { - m := NewEmbeddedETCD(t) + m := newEmbeddedETCD(t) ctx := t.Context() path := "/ident" @@ -73,7 +73,7 @@ func TestEphemeral(t *testing.T) { } func TestEphemeralFailedAsPutAlready(t *testing.T) { - m := NewEmbeddedETCD(t) + m := newEmbeddedETCD(t) ctx := t.Context() path := "/ident" diff --git a/store/etcdv3/meta/etcd.go b/store/etcdv3/meta/etcd.go index 931c86378..f578a6c55 100644 --- a/store/etcdv3/meta/etcd.go +++ b/store/etcdv3/meta/etcd.go @@ -202,7 +202,7 @@ func (e *ETCD) BindStatus(ctx context.Context, entityKey, statusKey, statusValue return e.bindStatusWithTTL(ctx, entityKey, statusKey, statusValue, ttl) } -func (e *ETCD) BatchCreateAndDecr(ctx context.Context, data map[string]string, decrKey string) (err error) { +func (e *ETCD) BatchCreateAndDecr(ctx context.Context, data map[string]string, decrKey string) error { resp, err := e.Get(ctx, decrKey) if err != nil { return err @@ -423,7 +423,7 @@ func (e *ETCD) doBatchOp(ctx context.Context, transactions []ETCDTxn) (*clientv3 } spans = append(spans, span{lastIdx, len(txnes)}) - // indexed slots keep the merged responses in request order, which GetMulti pairs with its keys + // Indexed slots keep the merged responses in request order, which GetMulti pairs with its keys resps := make([]*clientv3.TxnResponse, len(spans)) g, ctx := errgroup.WithContext(ctx) g.SetLimit(runtime.GOMAXPROCS(0)) diff --git a/store/etcdv3/meta/etcd_test.go b/store/etcdv3/meta/etcd_test.go index 01608baa6..da2d839d1 100644 --- a/store/etcdv3/meta/etcd_test.go +++ b/store/etcdv3/meta/etcd_test.go @@ -2,7 +2,7 @@ package meta import ( "context" - "fmt" + "errors" "strconv" "sync" "testing" @@ -19,8 +19,8 @@ import ( ) func TestGetOneError(t *testing.T) { - e := NewMockedETCD(t) - expErr := fmt.Errorf("exp") + e := newMockedETCD(t) + expErr := errors.New("exp") e.cliv3.(*mocks.ETCDClientV3).On("Get", mock.Anything, mock.Anything).Return(nil, expErr) kv, err := e.GetOne(t.Context(), "foo") require.Equal(t, expErr, err) @@ -28,7 +28,7 @@ func TestGetOneError(t *testing.T) { } func TestGetOneFailedAsRespondMore(t *testing.T) { - e := NewMockedETCD(t) + e := newMockedETCD(t) expResp := &clientv3.GetResponse{Count: 2} e.cliv3.(*mocks.ETCDClientV3).On("Get", mock.Anything, mock.Anything).Return(expResp, nil) kv, err := e.GetOne(t.Context(), "foo") @@ -37,7 +37,7 @@ func TestGetOneFailedAsRespondMore(t *testing.T) { } func TestGetOneMissingKeyIsNotFound(t *testing.T) { - e := NewEmbeddedETCD(t) + e := newEmbeddedETCD(t) kv, err := e.GetOne(t.Context(), "/absent") require.ErrorIs(t, err, types.ErrKeyNotFound) require.NotErrorIs(t, err, types.ErrInvaildCount) @@ -46,15 +46,15 @@ func TestGetOneMissingKeyIsNotFound(t *testing.T) { } func TestGetMultiWithNoKeys(t *testing.T) { - e := NewEmbeddedETCD(t) + e := newEmbeddedETCD(t) kvs, err := e.GetMulti(t.Context(), []string{}) require.NoError(t, err) require.Equal(t, 0, len(kvs)) } func TestGetMultiFailedAsBatchGetError(t *testing.T) { - e := NewMockedETCD(t) - expErr := fmt.Errorf("exp") + e := newMockedETCD(t) + expErr := errors.New("exp") expTxn := &mocks.Txn{} expTxn.On("If", mock.Anything).Return(expTxn) expTxn.On("Then", mock.Anything).Return(expTxn) @@ -67,7 +67,7 @@ func TestGetMultiFailedAsBatchGetError(t *testing.T) { } func TestGetMultiMissingKeyIsNotFound(t *testing.T) { - e := NewEmbeddedETCD(t) + e := newEmbeddedETCD(t) _, err := e.cliv3.Put(t.Context(), "/present", "v") require.NoError(t, err) kvs, err := e.GetMulti(t.Context(), []string{"/present", "/absent"}) @@ -76,19 +76,10 @@ func TestGetMultiMissingKeyIsNotFound(t *testing.T) { require.Nil(t, kvs) } -func TestGrant(t *testing.T) { - e := NewMockedETCD(t) - expErr := fmt.Errorf("exp") - e.cliv3.(*mocks.ETCDClientV3).On("Grant", mock.Anything, mock.Anything).Return(nil, expErr) - resp, err := e.cliv3.Grant(t.Context(), 1) - require.Equal(t, expErr, err) - require.Nil(t, resp) -} - func TestBindStatusFailedAsGrantError(t *testing.T) { e, etcd, assert := testKeepAliveETCD(t) defer assert() - expErr := fmt.Errorf("exp") + expErr := errors.New("exp") txn := &mocks.Txn{} defer txn.AssertExpectations(t) txn.On("If", mock.Anything, mock.Anything).Return(txn) @@ -104,7 +95,7 @@ func TestBindStatusFailedAsCommitError(t *testing.T) { e, etcd, assert := testKeepAliveETCD(t) defer assert() - expErr := fmt.Errorf("exp") + expErr := errors.New("exp") txn := &mocks.Txn{} defer txn.AssertExpectations(t) txn.On("If", mock.Anything, mock.Anything).Return(txn).Once() @@ -147,7 +138,7 @@ func TestBindStatusRenewsAnUnchangedStatus(t *testing.T) { etcd.On("Txn", mock.Anything).Return(txn) etcd.On("KeepAliveOnce", mock.Anything, clientv3.LeaseID(leaseID)).Return(&clientv3.LeaseKeepAliveResponse{TTL: 1}, nil) - require.Equal(t, nil, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) + require.NoError(t, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) etcd.AssertNotCalled(t, "Grant", mock.Anything, mock.Anything) } @@ -184,11 +175,11 @@ func TestBindStatusRebindsWhenTheTTLChanged(t *testing.T) { etcd.On("Txn", mock.Anything).Return(txn) etcd.On("KeepAliveOnce", mock.Anything, clientv3.LeaseID(leaseID)).Return(&clientv3.LeaseKeepAliveResponse{TTL: 5}, nil) etcd.On("Grant", mock.Anything, mock.Anything).Return(&clientv3.LeaseGrantResponse{}, nil) - require.Equal(t, nil, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) + require.NoError(t, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) } func TestBindStatusWithoutEntityCarriesALease(t *testing.T) { - e := NewEmbeddedETCD(t) + e := newEmbeddedETCD(t) ctx := t.Context() require.NoError(t, e.BindStatus(ctx, "/entity", "/status", "gone", 0)) @@ -248,7 +239,7 @@ func TestBindStatusWithZeroTTL(t *testing.T) { etcd.On("Txn", mock.Anything).Return(txn) - require.Equal(t, nil, e.BindStatus(t.Context(), "/entity", "/status", "status", 0)) + require.NoError(t, e.BindStatus(t.Context(), "/entity", "/status", "status", 0)) } func TestBindStatusRebindsAChangedValue(t *testing.T) { @@ -265,12 +256,12 @@ func TestBindStatusRebindsAChangedValue(t *testing.T) { etcd.On("Grant", mock.Anything, mock.Anything).Return(&clientv3.LeaseGrantResponse{}, nil) etcd.On("Txn", mock.Anything).Return(txn) - require.Equal(t, nil, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) + require.NoError(t, e.BindStatus(t.Context(), "/entity", "/status", "status", 1)) etcd.AssertNotCalled(t, "KeepAliveOnce", mock.Anything, mock.Anything) } func TestBindStatusKeepsOneLeaseAcrossRepeatedReports(t *testing.T) { - e := NewEmbeddedETCD(t) + e := newEmbeddedETCD(t) ctx := t.Context() _, err := e.Put(ctx, "/entity", "here") require.NoError(t, err) @@ -291,7 +282,7 @@ func TestBindStatusKeepsOneLeaseAcrossRepeatedReports(t *testing.T) { } func TestETCD(t *testing.T) { - m := NewEmbeddedETCD(t) + m := newEmbeddedETCD(t) ctx := t.Context() _, err := m.CreateLock("test", 5) @@ -348,18 +339,14 @@ func TestETCD(t *testing.T) { r, err = m.BatchUpdate(ctx, data) require.EqualError(t, err, "key not exists") require.False(t, r.Succeeded) - ctx2, cancel := context.WithCancel(ctx) - ch := m.Watch(ctx2, "watchkey", clientv3.WithPrefix()) - go func() { - for r := range ch { - require.NotEmpty(t, r.Events) - require.Equal(t, len(r.Events), 1) - require.Equal(t, r.Events[0].Type, clientv3.EventTypePut) - require.Equal(t, string(r.Events[0].Kv.Value), "b") - } - }() - m.Create(ctx, "watchkey/1", "b") + created, err := m.Create(ctx, "watchkey/1", "b") + require.NoError(t, err) + watchCtx, cancel := context.WithCancel(ctx) + watched := <-m.Watch(watchCtx, "watchkey", clientv3.WithPrefix(), clientv3.WithRev(created.Header.Revision)) cancel() + require.Len(t, watched.Events, 1) + require.Equal(t, clientv3.EventTypePut, watched.Events[0].Type) + require.Equal(t, "b", string(watched.Events[0].Kv.Value)) data = map[string]string{ "bcad_k1": "v1", @@ -449,7 +436,7 @@ func TestETCD(t *testing.T) { } func TestBatchCreateAndDecrWithTheCounterGoneMidRetry(t *testing.T) { - e := NewMockedETCD(t) + e := newMockedETCD(t) cli := e.cliv3.(*mocks.ETCDClientV3) cli.On("Get", mock.Anything, mock.Anything).Return(&clientv3.GetResponse{ Count: 1, @@ -470,13 +457,13 @@ func TestBatchCreateAndDecrWithTheCounterGoneMidRetry(t *testing.T) { require.ErrorIs(t, err, types.ErrKeyNotExists) } -func NewMockedETCD(t *testing.T) *ETCD { - e := NewEmbeddedETCD(t) +func newMockedETCD(t *testing.T) *ETCD { + e := newEmbeddedETCD(t) e.cliv3 = &mocks.ETCDClientV3{} return e } -func NewEmbeddedETCD(t *testing.T) *ETCD { +func newEmbeddedETCD(t *testing.T) *ETCD { config := types.EtcdConfig{ Machines: []string{"127.0.0.1:2379"}, Prefix: "/eru-test", @@ -491,7 +478,7 @@ func NewEmbeddedETCD(t *testing.T) *ETCD { } func testKeepAliveETCD(t *testing.T) (*ETCD, *mocks.ETCDClientV3, func()) { - e := NewMockedETCD(t) + e := newMockedETCD(t) etcd, ok := e.cliv3.(*mocks.ETCDClientV3) require.True(t, ok) return e, etcd, func() { etcd.AssertExpectations(t) } diff --git a/store/redis/rediaron.go b/store/redis/rediaron.go index 905264945..ddfdeb1c9 100644 --- a/store/redis/rediaron.go +++ b/store/redis/rediaron.go @@ -241,7 +241,7 @@ func (r *Rediaron) BindStatus(ctx context.Context, entityKey, statusKey, statusV if err != nil { return err } - // mirrors etcd: a missing entity key is an error for a status that carries a ttl + // Mirrors etcd: a missing entity key is an error for a status that carries a ttl if bound == replyMissing { return types.ErrInvaildCount } diff --git a/types/options.go b/types/options.go index 422bbdb73..6ae935aa7 100644 --- a/types/options.go +++ b/types/options.go @@ -4,7 +4,6 @@ import ( "cmp" "fmt" "io" - "slices" "strings" resourcetypes "github.com/projecteru2/core/resource/types" @@ -71,7 +70,6 @@ type DeployOptions struct { ExtraHosts []string Networks map[string]string // network name to specified IP User string - Debug bool // use syslog as log driver OpenStdin bool Labels map[string]string DeployStrategy string @@ -81,7 +79,6 @@ type DeployOptions struct { IgnoreHook bool AfterCreate []string RawArgs RawArgs - Lambda bool IgnorePull bool } @@ -138,17 +135,6 @@ type LinuxFile struct { Mode int64 } -// Clone deep-copies Content. -func (f LinuxFile) Clone() LinuxFile { - return LinuxFile{ - Content: slices.Clone(f.Content), - Filename: f.Filename, - UID: f.UID, - GID: f.GID, - Mode: f.Mode, - } -} - func (f LinuxFile) String() string { return fmt.Sprintf("file %+v:%+v:%+v:%#o, len: %+v", f.Filename, f.UID, f.GID, f.Mode, len(f.Content)) } diff --git a/types/options_test.go b/types/options_test.go index 09d18447c..082527752 100644 --- a/types/options_test.go +++ b/types/options_test.go @@ -2,10 +2,10 @@ package types import ( "bufio" + "bytes" "testing" "github.com/stretchr/testify/assert" - "go.uber.org/zap/buffer" resourcetypes "github.com/projecteru2/core/resource/types" ) @@ -26,7 +26,7 @@ func TestDeployOptions(t *testing.T) { o := &DeployOptions{Entrypoint: &Entrypoint{}} assert.Equal(ErrEmptyAppName, o.Validate()) - assert.NotNil(t, o.GetProcessing("t")) + assert.NotNil(o.GetProcessing("t")) o.Name = "testname" assert.Equal(ErrEmptyPodName, o.Validate()) @@ -72,10 +72,8 @@ func TestCopyOptions(t *testing.T) { func TestLinuxFile(t *testing.T) { lf := LinuxFile{Filename: "s"} - nlf := lf.Clone() - assert.Equal(t, lf.Filename, nlf.Filename) assert.NotEmpty(t, lf.String()) - b := bufio.NewWriter(&buffer.Buffer{}) + b := bufio.NewWriter(&bytes.Buffer{}) lf.LitterDump(b) assert.NoError(t, b.Flush()) } @@ -168,10 +166,10 @@ func TestImageOptions(t *testing.T) { assert.NoError(o.Validate()) } -func TestRawArges(t *testing.T) { +func TestRawArgs(t *testing.T) { ra := RawArgs([]byte("abc")) assert.Equal(t, ra.String(), "abc") - b := bufio.NewWriter(&buffer.Buffer{}) + b := bufio.NewWriter(&bytes.Buffer{}) ra.LitterDump(b) assert.NoError(t, b.Flush()) } diff --git a/types/workload.go b/types/workload.go index 1128c397a..f0f096a02 100644 --- a/types/workload.go +++ b/types/workload.go @@ -19,7 +19,7 @@ type StatusMeta struct { Healthy bool `json:"healthy,omitempty"` Extension []byte `json:"extension,omitempty"` - // set only when writing workload status + // Set only when writing workload status Appname string `json:"-"` Nodename string `json:"-"` Entrypoint string `json:"-"` diff --git a/utils/config.go b/utils/config.go index ef8b5c6d2..c3632188b 100644 --- a/utils/config.go +++ b/utils/config.go @@ -33,7 +33,7 @@ func LoadConfig(configPath string) (types.Config, error) { return config, checkRequired(value) } -// defaults land before the file is read so an explicit zero in the file still wins +// Defaults land before the file is read so an explicit zero in the file still wins func applyDefaults(value reflect.Value) error { return walkFields(value, func(field reflect.Value, structField reflect.StructField) error { if tag := structField.Tag.Get("default"); tag != "" && field.IsZero() { diff --git a/utils/transaction.go b/utils/transaction.go index 713c9adc6..acbbadc79 100644 --- a/utils/transaction.go +++ b/utils/transaction.go @@ -47,7 +47,7 @@ func Txn(ctx context.Context, cond, then contextFunc, rollback rollbackFunc, ttl condErr = cond(txnCtx) } if condErr == nil && then != nil { - // with no rollback, then must not be interruptible + // With no rollback, then must not be interruptible thenCtx := txnCtx var thenCancel context.CancelFunc if rollback == nil { diff --git a/wal/hydro.go b/wal/hydro.go index 13001ff7f..9d6488c68 100644 --- a/wal/hydro.go +++ b/wal/hydro.go @@ -35,7 +35,7 @@ type Hydro struct { } func NewHydro(ctx context.Context, store Store, address string, config coretypes.Config) (*Hydro, error) { - // the journal outlives every request that writes to it, so it keeps a context of its own + // The journal outlives every request that writes to it, so it keeps a context of its own hydro := &Hydro{handlers: map[string]EventHandler{}, store: store, ctx: context.WithoutCancel(ctx), config: config, address: address} seq, err := hydro.lastSeq(ctx) if err != nil {