diff --git a/persist/sqldb/explosive_offload_node_status_repo.go b/persist/sqldb/explosive_offload_node_status_repo.go index 6123644bda0b..4bb45a4e9839 100644 --- a/persist/sqldb/explosive_offload_node_status_repo.go +++ b/persist/sqldb/explosive_offload_node_status_repo.go @@ -26,7 +26,7 @@ func (n *explosiveOffloadNodeStatusRepo) Get(context.Context, string, string) (w return nil, ErrOffloadNotSupported } -func (n *explosiveOffloadNodeStatusRepo) List(context.Context, string) (map[UUIDVersion]wfv1.Nodes, error) { +func (n *explosiveOffloadNodeStatusRepo) List(context.Context, string, []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error) { return nil, ErrOffloadNotSupported } diff --git a/persist/sqldb/mocks/OffloadNodeStatusRepo.go b/persist/sqldb/mocks/OffloadNodeStatusRepo.go index c4d423fc9531..d68d6bfa98f4 100644 --- a/persist/sqldb/mocks/OffloadNodeStatusRepo.go +++ b/persist/sqldb/mocks/OffloadNodeStatusRepo.go @@ -67,13 +67,13 @@ func (_m *OffloadNodeStatusRepo) IsEnabled() bool { return r0 } -// List provides a mock function with given fields: namespace -func (_m *OffloadNodeStatusRepo) List(ctx context.Context, namespace string) (map[sqldb.UUIDVersion]v1alpha1.Nodes, error) { - ret := _m.Called(namespace) +// List provides a mock function with given fields: namespace, keys +func (_m *OffloadNodeStatusRepo) List(ctx context.Context, namespace string, keys []sqldb.UUIDVersion) (map[sqldb.UUIDVersion]v1alpha1.Nodes, error) { + ret := _m.Called(namespace, keys) var r0 map[sqldb.UUIDVersion]v1alpha1.Nodes - if rf, ok := ret.Get(0).(func(string) map[sqldb.UUIDVersion]v1alpha1.Nodes); ok { - r0 = rf(namespace) + if rf, ok := ret.Get(0).(func(string, []sqldb.UUIDVersion) map[sqldb.UUIDVersion]v1alpha1.Nodes); ok { + r0 = rf(namespace, keys) } else { if ret.Get(0) != nil { r0 = ret.Get(0).(map[sqldb.UUIDVersion]v1alpha1.Nodes) @@ -81,8 +81,8 @@ func (_m *OffloadNodeStatusRepo) List(ctx context.Context, namespace string) (ma } var r1 error - if rf, ok := ret.Get(1).(func(string) error); ok { - r1 = rf(namespace) + if rf, ok := ret.Get(1).(func(string, []sqldb.UUIDVersion) error); ok { + r1 = rf(namespace, keys) } else { r1 = ret.Error(1) } diff --git a/persist/sqldb/offload_node_status_repo.go b/persist/sqldb/offload_node_status_repo.go index 6a0218bc4213..1e91e9913906 100644 --- a/persist/sqldb/offload_node_status_repo.go +++ b/persist/sqldb/offload_node_status_repo.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "hash/fnv" + "slices" "strings" "time" @@ -16,8 +17,6 @@ import ( "github.com/argoproj/argo-workflows/v4/util/sqldb" ) -const OffloadNodeStatusDisabled = "Workflow has offloaded nodes, but offloading has been disabled" - type UUIDVersion struct { UID string `db:"uid"` Version string `db:"version"` @@ -26,7 +25,7 @@ type UUIDVersion struct { type OffloadNodeStatusRepo interface { Save(ctx context.Context, uid, namespace string, nodes wfv1.Nodes) (string, error) Get(ctx context.Context, uid, version string) (wfv1.Nodes, error) - List(ctx context.Context, namespace string) (map[UUIDVersion]wfv1.Nodes, error) + List(ctx context.Context, namespace string, keys []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error) ListOldOffloads(ctx context.Context, namespace string) (map[string][]string, error) Delete(ctx context.Context, uid, version string) error IsEnabled() bool @@ -149,34 +148,59 @@ func (wdc *nodeOffloadRepo) Get(ctx context.Context, uid, version string) (wfv1. return nodes, nil } -func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string) (map[UUIDVersion]wfv1.Nodes, error) { - wdc.log.WithFields(logging.Fields{"namespace": namespace}).Debug(ctx, "Listing offloaded nodes") - var res map[UUIDVersion]wfv1.Nodes - err := wdc.sessionProxy.With(ctx, func(s db.Session) error { - var records []nodesRecord - err := s.SQL(). - Select("uid", "version", "nodes"). - From(wdc.tableName). - Where(db.Cond{"clustername": wdc.clusterName}). - And(namespaceEqual(namespace)). - All(&records) - if err != nil { - return err - } +// offloadListBatchSize caps how many keys go into a single query. MySQL and Postgres allow +// 65535 placeholders per prepared statement and each pair costs two, so this leaves plenty of +// headroom. A ceiling is needed because the caller decides the key count: ListWorkflows treats +// an unset limit as "the whole namespace", and `argo list` defaults --chunk-size to 0. +const offloadListBatchSize = 1000 - res = make(map[UUIDVersion]wfv1.Nodes) - for _, r := range records { - nodes := &wfv1.Nodes{} - err = json.Unmarshal([]byte(r.Nodes), nodes) +// uuidVersionIn matches exactly the given (uid, version) pairs. Written as OR-of-ANDs rather +// than a row value `(uid, version) IN ((?,?),...)` so that it behaves the same on MySQL, +// MariaDB and Postgres. +func uuidVersionIn(keys []UUIDVersion) db.LogicalExpr { + conds := make([]db.LogicalExpr, len(keys)) + for i, key := range keys { + conds[i] = db.And(db.Cond{"uid": key.UID}, db.Cond{"version": key.Version}) + } + return db.Or(conds...) +} + +// List returns the offloaded nodes for the given keys only. +func (wdc *nodeOffloadRepo) List(ctx context.Context, namespace string, keys []UUIDVersion) (map[UUIDVersion]wfv1.Nodes, error) { + // This is not merely an optimisation: db.Or() with no arguments is an empty condition, + // so the query would silently widen back to every nodes blob in the namespace. + if len(keys) == 0 { + return map[UUIDVersion]wfv1.Nodes{}, nil + } + wdc.log.WithFields(logging.Fields{"namespace": namespace, "keys": len(keys)}).Debug(ctx, "Listing offloaded nodes") + res := make(map[UUIDVersion]wfv1.Nodes) + for batch := range slices.Chunk(keys, offloadListBatchSize) { + err := wdc.sessionProxy.With(ctx, func(s db.Session) error { + var records []nodesRecord + err := s.SQL(). + Select("uid", "version", "nodes"). + From(wdc.tableName). + Where(db.Cond{"clustername": wdc.clusterName}). + And(namespaceEqual(namespace)). + And(uuidVersionIn(batch)). + All(&records) if err != nil { return err } - res[UUIDVersion{UID: r.UID, Version: r.Version}] = *nodes + + for _, r := range records { + nodes := &wfv1.Nodes{} + err = json.Unmarshal([]byte(r.Nodes), nodes) + if err != nil { + return err + } + res[UUIDVersion{UID: r.UID, Version: r.Version}] = *nodes + } + return nil + }) + if err != nil { + return nil, err } - return nil - }) - if err != nil { - return nil, err } return res, nil } diff --git a/persist/sqldb/offload_node_status_repo_mysql_test.go b/persist/sqldb/offload_node_status_repo_mysql_test.go new file mode 100644 index 000000000000..ca109d6d5145 --- /dev/null +++ b/persist/sqldb/offload_node_status_repo_mysql_test.go @@ -0,0 +1,72 @@ +//go:build !windows + +package sqldb + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/argoproj/argo-workflows/v4/util/logging" + usqldb "github.com/argoproj/argo-workflows/v4/util/sqldb" +) + +// setupMySQLOffloadTest starts a MySQL or MariaDB container and returns an offload repository. +func setupMySQLOffloadTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) OffloadNodeStatusRepo { + t.Helper() + repo, err := NewOffloadNodeStatusRepo(ctx, logging.RequireLoggerFromContext(ctx), setupMySQLTest(ctx, t, v), "test", "argo_workflows") + require.NoError(t, err) + return repo +} + +// saveOffload writes a node status and returns the version it was stored under. +func saveOffload(ctx context.Context, t *testing.T, repo OffloadNodeStatusRepo, uid, nodeName string) string { + t.Helper() + version, err := repo.Save(ctx, uid, "argo", wfv1.Nodes{nodeName: wfv1.NodeStatus{ID: nodeName}}) + require.NoError(t, err) + return version +} + +// TestMySQLListOnlyReturnsRequestedKeys covers the behaviour the list path depends on: the +// query matches whole (uid, version) pairs, so a caller that needs one page of workflows does +// not pull every offloaded blob in the namespace. +// +// Save deliberately leaves superseded rows behind for the garbage collector, so uid-a and uid-b +// each have two versions here and only one version of each is requested. +// +// The version is a hash of the node contents, so two workflows only share a version value when +// they store identical nodes. That is arranged deliberately: without it every version value is +// unique to one uid, and a `uid IN (...) AND version IN (...)` cross-product happens to return +// the right rows anyway. With shared version values the cross-product matches all four rows, +// while matching whole pairs returns two. +func TestMySQLListOnlyReturnsRequestedKeys(t *testing.T) { + for name, variant := range usqldb.MySQLVariants { + t.Run(name, func(t *testing.T) { + ctx := logging.TestContext(t.Context()) + repo := setupMySQLOffloadTest(ctx, t, variant) + + versionX := saveOffload(ctx, t, repo, "uid-a", "node-x") + versionY := saveOffload(ctx, t, repo, "uid-a", "node-y") + require.NotEqual(t, versionX, versionY, "different nodes must produce different versions") + require.Equal(t, versionX, saveOffload(ctx, t, repo, "uid-b", "node-x"), "identical nodes must share a version") + require.Equal(t, versionY, saveOffload(ctx, t, repo, "uid-b", "node-y"), "identical nodes must share a version") + unwanted := UUIDVersion{UID: "uid-c", Version: saveOffload(ctx, t, repo, "uid-c", "node-c")} + + wantedA := UUIDVersion{UID: "uid-a", Version: versionX} + wantedB := UUIDVersion{UID: "uid-b", Version: versionY} + + got, err := repo.List(ctx, "argo", []UUIDVersion{wantedA, wantedB}) + require.NoError(t, err) + + assert.Len(t, got, 2) + assert.Contains(t, got, wantedA) + assert.Contains(t, got, wantedB) + assert.NotContains(t, got, UUIDVersion{UID: "uid-a", Version: versionY}, "a version that was not asked for must not come back") + assert.NotContains(t, got, UUIDVersion{UID: "uid-b", Version: versionX}, "a version that was not asked for must not come back") + assert.NotContains(t, got, unwanted, "a uid that was not asked for must not come back") + }) + } +} diff --git a/persist/sqldb/offload_node_status_repo_test.go b/persist/sqldb/offload_node_status_repo_test.go index 6a39284dcaed..1efeeb51ddfc 100644 --- a/persist/sqldb/offload_node_status_repo_test.go +++ b/persist/sqldb/offload_node_status_repo_test.go @@ -7,8 +7,18 @@ import ( "github.com/stretchr/testify/require" wfv1 "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + "github.com/argoproj/argo-workflows/v4/util/logging" ) +// Test_ListWithoutKeys guards the empty-keys early return. Without it the generated condition +// is empty, which would widen the query back to the whole namespace. A zero-value repo has no +// session, so anything past the guard panics rather than quietly querying. +func Test_ListWithoutKeys(t *testing.T) { + got, err := (&nodeOffloadRepo{}).List(logging.TestContext(t.Context()), "argo", nil) + require.NoError(t, err) + assert.Empty(t, got) +} + func Test_nodeStatusVersion(t *testing.T) { t.Run("Empty", func(t *testing.T) { marshalled, version, err := nodeStatusVersion(nil) diff --git a/persist/sqldb/workflow_archive_mysql_test.go b/persist/sqldb/workflow_archive_mysql_test.go index eabf96828d39..1b3759bbbdfc 100644 --- a/persist/sqldb/workflow_archive_mysql_test.go +++ b/persist/sqldb/workflow_archive_mysql_test.go @@ -27,6 +27,13 @@ import ( // setupMySQLArchiveTest starts a MySQL or MariaDB container, runs migrations, and returns a WorkflowArchive. func setupMySQLArchiveTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) WorkflowArchive { t.Helper() + return NewWorkflowArchive(setupMySQLTest(ctx, t, v), "test", "", instanceid.NewService("")) +} + +// setupMySQLTest starts a MySQL or MariaDB container, runs migrations, and returns a session +// proxy the caller can build any repository on. +func setupMySQLTest(ctx context.Context, t *testing.T, v usqldb.MySQLVariant) *usqldb.SessionProxy { + t.Helper() c, err := testmysql.Run(ctx, v.Image, @@ -73,7 +80,7 @@ func setupMySQLArchiveTest(ctx context.Context, t *testing.T, v usqldb.MySQLVari t.Cleanup(func() { proxy.Close() }) - return NewWorkflowArchive(proxy, "test", "", instanceid.NewService("")) + return proxy } // TestMySQLListWorkflows verifies that JSON_EXTRACT/JSON_UNQUOTE queries in diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go index efa016e93a2b..e4187d112aed 100644 --- a/server/workflow/workflow_server.go +++ b/server/workflow/workflow_server.go @@ -5,6 +5,8 @@ import ( "encoding/json" "fmt" "io" + "maps" + "slices" "sort" "sync" "time" @@ -285,19 +287,22 @@ func (s *workflowServer) ListWorkflows(ctx context.Context, req *workflowpkg.Wor } cleaner := fields.NewCleaner(req.Fields) - logger := logging.RequireLoggerFromContext(ctx) if s.offloadNodeStatusRepo.IsEnabled() && !cleaner.WillExclude("items.status.nodes") { - offloadedNodes, err := s.offloadNodeStatusRepo.List(ctx, req.Namespace) - if err != nil { - return nil, sutils.ToStatusError(err, codes.Internal) - } + // This page is already resolved, so we know exactly which offloaded rows we need. + offloaded := map[int]sqldb.UUIDVersion{} for i, wf := range wfs { if wf.Status.IsOffloadNodeStatus() { - if s.offloadNodeStatusRepo.IsEnabled() { - wfs[i].Status.Nodes = offloadedNodes[sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()}] - } else { - logger.WithFields(logging.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(ctx, sqldb.OffloadNodeStatusDisabled) - } + offloaded[i] = sqldb.UUIDVersion{UID: string(wf.UID), Version: wf.GetOffloadNodeStatusVersion()} + } + } + // Nothing on this page is offloaded, so there is nothing to fetch. + if len(offloaded) > 0 { + offloadedNodes, err := s.offloadNodeStatusRepo.List(ctx, req.Namespace, slices.Collect(maps.Values(offloaded))) + if err != nil { + return nil, sutils.ToStatusError(err, codes.Internal) + } + for i, key := range offloaded { + wfs[i].Status.Nodes = offloadedNodes[key] } } } diff --git a/server/workflow/workflow_server_offload_test.go b/server/workflow/workflow_server_offload_test.go new file mode 100644 index 000000000000..d3f056532103 --- /dev/null +++ b/server/workflow/workflow_server_offload_test.go @@ -0,0 +1,126 @@ +package workflow + +import ( + "context" + "testing" + "time" + + "github.com/go-jose/go-jose/v4/jwt" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + "github.com/argoproj/argo-workflows/v4/persist/sqldb" + "github.com/argoproj/argo-workflows/v4/persist/sqldb/mocks" + workflowpkg "github.com/argoproj/argo-workflows/v4/pkg/apiclient/workflow" + "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" + v1alpha "github.com/argoproj/argo-workflows/v4/pkg/client/clientset/versioned/fake" + authtypes "github.com/argoproj/argo-workflows/v4/server/auth/types" + "github.com/argoproj/argo-workflows/v4/server/workflow/store" + "github.com/argoproj/argo-workflows/v4/util/instanceid" +) + +// offloadedWorkflow builds a workflow whose node status lives in the offload table. Pages are +// ordered by startedat descending, so startedAt decides which page a fixture lands on. +func offloadedWorkflow(name, uid, version string, startedAt time.Time) v1alpha1.Workflow { + return v1alpha1.Workflow{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "argo", UID: types.UID(uid)}, + Status: v1alpha1.WorkflowStatus{ + OffloadNodeStatusVersion: version, + StartedAt: metav1.NewTime(startedAt), + }, + } +} + +// inlineWorkflow builds a workflow that keeps its node status on the object itself, so the +// offload table has nothing for it. +func inlineWorkflow(name, uid string) v1alpha1.Workflow { + return offloadedWorkflow(name, uid, "", time.Time{}) +} + +// offloadTestServer builds the smallest server that can serve ListWorkflows, and hands back +// the offload mock so tests can assert on how it was called. +func offloadTestServer(t *testing.T, wfs ...v1alpha1.Workflow) (Server, context.Context, *mocks.OffloadNodeStatusRepo) { + t.Helper() + + offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{} + offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true) + + archivedRepo := &mocks.WorkflowArchive{} + archivedRepo.On("CountWorkflows", mock.Anything, mock.Anything).Return(int64(0), nil) + archivedRepo.On("ListWorkflows", mock.Anything, mock.Anything).Return(v1alpha1.Workflows{}, nil) + archivedRepo.On("HasMoreWorkflows", mock.Anything, mock.Anything).Return(false, nil) + + // An empty instance ID keeps the store from requiring an instance-id label on the fixtures. + instanceIDSvc := instanceid.NewService("") + wfStore, err := store.NewSQLiteStore(instanceIDSvc) + require.NoError(t, err) + for i := range wfs { + require.NoError(t, wfStore.Add(&wfs[i])) + } + + server, ctx := newTestServer(t, testServerOpts{ + instanceIDSvc: instanceIDSvc, + offloadRepo: offloadNodeStatusRepo, + archivedRepo: archivedRepo, + wfClientset: v1alpha.NewClientset(), + wfLister: wfStore, + namespace: "argo", + claims: &authtypes.Claims{Claims: jwt.Claims{Subject: "my-sub"}}, + }) + return server, ctx, offloadNodeStatusRepo +} + +// TestListWorkflows_PassesOnlyPageKeys asserts that the offload query is scoped to the +// workflows on this page, and that what comes back is attached to the right workflow. +func TestListWorkflows_PassesOnlyPageKeys(t *testing.T) { + now := time.Now() + first := offloadedWorkflow("offloaded-first", "uid-first", "v1", now) + second := offloadedWorkflow("offloaded-second", "uid-second", "v2", now.Add(-time.Minute)) + // Offloaded as well, but a Limit of 2 puts it on the next page. + nextPage := offloadedWorkflow("offloaded-next-page", "uid-next-page", "v3", now.Add(-2*time.Minute)) + server, ctx, offloadNodeStatusRepo := offloadTestServer(t, first, second, nextPage) + + secondNodes := v1alpha1.Nodes{"n": v1alpha1.NodeStatus{ID: "n", Name: "belongs-to-second"}} + offloadNodeStatusRepo.On("List", mock.Anything, mock.Anything). + Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{{UID: "uid-second", Version: "v2"}: secondNodes}, nil) + + list, err := server.ListWorkflows(ctx, &workflowpkg.WorkflowListRequest{ + Namespace: "argo", + ListOptions: &metav1.ListOptions{Limit: 2}, + }) + require.NoError(t, err) + + offloadNodeStatusRepo.AssertNumberOfCalls(t, "List", 1) + call := offloadNodeStatusRepo.Calls[len(offloadNodeStatusRepo.Calls)-1] + assert.Equal(t, "argo", call.Arguments[0]) + assert.ElementsMatch(t, []sqldb.UUIDVersion{ + {UID: "uid-first", Version: "v1"}, + {UID: "uid-second", Version: "v2"}, + }, call.Arguments[1], "List must be given exactly the offloaded keys on this page") + + // The page is sorted before it is returned, so look the fixtures up by name rather than + // by position. + require.Len(t, list.Items, 2) + nodesByName := map[string]v1alpha1.Nodes{} + for _, wf := range list.Items { + nodesByName[wf.Name] = wf.Status.Nodes + } + assert.Equal(t, secondNodes, nodesByName["offloaded-second"], "nodes must land on the workflow they were keyed by") + assert.Empty(t, nodesByName["offloaded-first"], "a workflow with no offloaded row must not pick up another one's nodes") +} + +// TestListWorkflows_SkipsQueryWhenPageHasNoOffload asserts that a page with nothing offloaded +// issues no offload query at all. This is the common case for most users. +func TestListWorkflows_SkipsQueryWhenPageHasNoOffload(t *testing.T) { + server, ctx, offloadNodeStatusRepo := offloadTestServer(t, inlineWorkflow("inline-a", "uid-a"), inlineWorkflow("inline-b", "uid-b")) + + _, err := server.ListWorkflows(ctx, &workflowpkg.WorkflowListRequest{Namespace: "argo"}) + require.NoError(t, err) + + // AssertNotCalled without argument matchers can never fail: it looks for a recorded call + // matching the (empty) argument list it was given, which no real two-argument call does. + offloadNodeStatusRepo.AssertNumberOfCalls(t, "List", 0) +} diff --git a/server/workflow/workflow_server_test.go b/server/workflow/workflow_server_test.go index 372cc04dca6e..7cfe6cf96d5e 100644 --- a/server/workflow/workflow_server_test.go +++ b/server/workflow/workflow_server_test.go @@ -578,6 +578,43 @@ const clusterworkflowtmpl = ` const userEmailLabel = "my-sub.at.your.org" +// testServerOpts collects the parts the harnesses in this package actually vary. Everything +// they share — the selfsubjectaccessreviews reactor, the auth context chain, the template +// stores and the 12-argument NewServer call — lives in newTestServer. +type testServerOpts struct { + instanceIDSvc instanceid.Service + offloadRepo sqldb.OffloadNodeStatusRepo + archivedRepo sqldb.WorkflowArchive + wfClientset versioned.Interface + wfLister store.WorkflowLister + wfStore store.WorkflowStore + namespace string + claims *types.Claims + artifactRepos artifactrepositories.Interface +} + +// newTestServer builds a workflow server backed by fakes. Callers own the mocks they pass in, +// so a test that needs to assert on one keeps its own reference. +func newTestServer(t *testing.T, o testServerOpts) (Server, context.Context) { + t.Helper() + + kubeClientSet := fake.NewClientset() + kubeClientSet.PrependReactor("create", "selfsubjectaccessreviews", func(action ktesting.Action) (handled bool, ret runtime.Object, err error) { + return true, &authorizationv1.SelfSubjectAccessReview{ + Status: authorizationv1.SubjectAccessReviewStatus{Allowed: true}, + }, nil + }) + + ctx := logging.TestContext(t.Context()) + ctx = context.WithValue(ctx, auth.WfKey, o.wfClientset) + ctx = context.WithValue(ctx, auth.KubeKey, kubeClientSet) + ctx = context.WithValue(ctx, auth.ClaimsKey, o.claims) + + server := NewServer(ctx, o.instanceIDSvc, o.offloadRepo, o.archivedRepo, o.wfClientset, o.wfLister, o.wfStore, + workflowtemplate.NewClientStore(), clusterworkflowtemplate.NewClientStore(), nil, &o.namespace, o.artifactRepos) + return server, ctx +} + func getWorkflowServer(t *testing.T) (workflowpkg.WorkflowServiceServer, context.Context) { t.Helper() var unlabelledObj, wfObj1, wfObj2, wfObj3, wfObj4, wfObj5, failedWfObj v1alpha1.Workflow @@ -598,7 +635,7 @@ func getWorkflowServer(t *testing.T) (workflowpkg.WorkflowServiceServer, context offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{} offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true) - offloadNodeStatusRepo.On("List", mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil) + offloadNodeStatusRepo.On("List", mock.Anything, mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil) archivedRepo := &mocks.WorkflowArchive{} @@ -637,16 +674,8 @@ func getWorkflowServer(t *testing.T) (workflowpkg.WorkflowServiceServer, context archivedRepo.On("ListWorkflows", mock.Anything, sutils.ListOptions{Namespace: "test", Limit: -1, LabelRequirements: r}).Return(v1alpha1.Workflows{wfObj4}, nil) archivedRepo.On("HasMoreWorkflows", mock.Anything, sutils.ListOptions{Namespace: "test", LabelRequirements: r}).Return(false, nil) - kubeClientSet := fake.NewClientset() - kubeClientSet.PrependReactor("create", "selfsubjectaccessreviews", func(action ktesting.Action) (handled bool, ret runtime.Object, err error) { - return true, &authorizationv1.SelfSubjectAccessReview{ - Status: authorizationv1.SubjectAccessReviewStatus{Allowed: true}, - }, nil - }) wfClientset := v1alpha.NewClientset(&unlabelledObj, &wfObj1, &wfObj2, &wfObj3, &wfObj4, &wfObj5, &failedWfObj, &wftmpl, &cronwfObj, &cwfTmpl) wfClientset.PrependReactor("create", "workflows", generateNameReactor) - ctx := logging.TestContext(t.Context()) - ctx = context.WithValue(context.WithValue(context.WithValue(ctx, auth.WfKey, wfClientset), auth.KubeKey, kubeClientSet), auth.ClaimsKey, &types.Claims{Claims: jwt.Claims{Subject: "my-sub"}, Email: "my-sub@your.org"}) listOptions := &metav1.ListOptions{} instanceIDSvc := instanceid.NewService("my-instanceid") instanceIDSvc.With(listOptions) @@ -663,11 +692,16 @@ func getWorkflowServer(t *testing.T) (workflowpkg.WorkflowServiceServer, context if err = wfStore.Add(&wfObj5); err != nil { panic(err) } - namespaceAll := metav1.NamespaceAll - wftmplStore := workflowtemplate.NewClientStore() - cwftmplStore := clusterworkflowtemplate.NewClientStore() - server := NewServer(ctx, instanceIDSvc, offloadNodeStatusRepo, archivedRepo, wfClientset, wfStore, wfStore, wftmplStore, cwftmplStore, nil, &namespaceAll, nil) - return server, ctx + return newTestServer(t, testServerOpts{ + instanceIDSvc: instanceIDSvc, + offloadRepo: offloadNodeStatusRepo, + archivedRepo: archivedRepo, + wfClientset: wfClientset, + wfLister: wfStore, + wfStore: wfStore, + namespace: metav1.NamespaceAll, + claims: &types.Claims{Claims: jwt.Claims{Subject: "my-sub"}, Email: "my-sub@your.org"}, + }) } // generateNameReactor implements the logic required for the GenerateName field to work when using @@ -1501,38 +1535,31 @@ func getWorkflowServerWithArtifacts(t *testing.T, template runtime.Object, defau offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{} offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true) - offloadNodeStatusRepo.On("List", mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil) + offloadNodeStatusRepo.On("List", mock.Anything, mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil) archivedRepo := &mocks.WorkflowArchive{} - kubeClientSet := fake.NewSimpleClientset() - kubeClientSet.PrependReactor("create", "selfsubjectaccessreviews", func(action ktesting.Action) (handled bool, ret runtime.Object, err error) { - return true, &authorizationv1.SelfSubjectAccessReview{ - Status: authorizationv1.SubjectAccessReviewStatus{Allowed: true}, - }, nil - }) - wfClientset := v1alpha.NewClientset(template) wfClientset.PrependReactor("create", "workflows", generateNameReactor) - ctx := logging.TestContext(t.Context()) - ctx = context.WithValue(ctx, auth.WfKey, wfClientset) - ctx = context.WithValue(ctx, auth.KubeKey, kubeClientSet) - ctx = context.WithValue(ctx, auth.ClaimsKey, &types.Claims{Claims: jwt.Claims{Subject: "my-sub"}}) - - wfStore, err := store.NewSQLiteStore(instanceid.NewService("my-instanceid")) + instanceIDSvc := instanceid.NewService("my-instanceid") + wfStore, err := store.NewSQLiteStore(instanceIDSvc) require.NoError(t, err) - wftmplStore := workflowtemplate.NewClientStore() - cwftmplStore := clusterworkflowtemplate.NewClientStore() - var artifactRepos artifactrepositories.Interface if defaultRepo != nil { artifactRepos = armocks.DummyArtifactRepositories(defaultRepo) } - namespaceAll := metav1.NamespaceAll - server := NewServer(ctx, instanceid.NewService("my-instanceid"), offloadNodeStatusRepo, archivedRepo, wfClientset, wfStore, wfStore, wftmplStore, cwftmplStore, nil, &namespaceAll, artifactRepos) - - return server, ctx + return newTestServer(t, testServerOpts{ + instanceIDSvc: instanceIDSvc, + offloadRepo: offloadNodeStatusRepo, + archivedRepo: archivedRepo, + wfClientset: wfClientset, + wfLister: wfStore, + wfStore: wfStore, + namespace: metav1.NamespaceAll, + claims: &types.Claims{Claims: jwt.Claims{Subject: "my-sub"}}, + artifactRepos: artifactRepos, + }) } diff --git a/server/workflowarchive/archived_workflow_server_test.go b/server/workflowarchive/archived_workflow_server_test.go index cc6d823b1479..f867bdfee487 100644 --- a/server/workflowarchive/archived_workflow_server_test.go +++ b/server/workflowarchive/archived_workflow_server_test.go @@ -19,7 +19,6 @@ import ( kubefake "k8s.io/client-go/kubernetes/fake" k8stesting "k8s.io/client-go/testing" - "github.com/argoproj/argo-workflows/v4/persist/sqldb" "github.com/argoproj/argo-workflows/v4/persist/sqldb/mocks" workflowarchivepkg "github.com/argoproj/argo-workflows/v4/pkg/apiclient/workflowarchive" "github.com/argoproj/argo-workflows/v4/pkg/apis/workflow/v1alpha1" @@ -35,7 +34,6 @@ func Test_archivedWorkflowServer(t *testing.T) { wfClient := &argofake.Clientset{} offloadNodeStatusRepo := &mocks.OffloadNodeStatusRepo{} offloadNodeStatusRepo.On("IsEnabled", mock.Anything).Return(true) - offloadNodeStatusRepo.On("List", mock.Anything).Return(map[sqldb.UUIDVersion]v1alpha1.Nodes{}, nil) w := NewWorkflowArchiveServer(repo, offloadNodeStatusRepo, nil) allowed := true kubeClient.AddReactor("create", "selfsubjectaccessreviews", func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {