From cfff7d1bfabafaea8cafdc1a71d2b157603166b2 Mon Sep 17 00:00:00 2001 From: Ying Li Date: Wed, 21 Jun 2017 16:06:50 -0700 Subject: [PATCH] Change the `Restore` action on store objects so that objects that already exist are updated, rather than everything being deleted and re-created. Signed-off-by: Ying Li --- manager/state/store/clusters.go | 18 +- manager/state/store/configs.go | 18 +- manager/state/store/extensions.go | 26 +- manager/state/store/memory_test.go | 545 ++++++++++++++++++++++++++++- manager/state/store/networks.go | 18 +- manager/state/store/nodes.go | 18 +- manager/state/store/object.go | 43 +++ manager/state/store/resources.go | 26 +- manager/state/store/secrets.go | 18 +- manager/state/store/services.go | 18 +- manager/state/store/tasks.go | 18 +- 11 files changed, 634 insertions(+), 132 deletions(-) diff --git a/manager/state/store/clusters.go b/manager/state/store/clusters.go index a495dacc92..495fc04037 100644 --- a/manager/state/store/clusters.go +++ b/manager/state/store/clusters.go @@ -43,21 +43,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - clusters, err := FindClusters(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Clusters)) + for i, x := range snapshot.Clusters { + toStoreObj[i] = x } - for _, n := range clusters { - if err := DeleteCluster(tx, n.ID); err != nil { - return err - } - } - for _, n := range snapshot.Clusters { - if err := CreateCluster(tx, n); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableCluster, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/configs.go b/manager/state/store/configs.go index 2bd44ae1bd..d02e04ba9b 100644 --- a/manager/state/store/configs.go +++ b/manager/state/store/configs.go @@ -37,21 +37,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - configs, err := FindConfigs(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Configs)) + for i, x := range snapshot.Configs { + toStoreObj[i] = x } - for _, s := range configs { - if err := DeleteConfig(tx, s.ID); err != nil { - return err - } - } - for _, s := range snapshot.Configs { - if err := CreateConfig(tx, s); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableConfig, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/extensions.go b/manager/state/store/extensions.go index 627d87d531..8dac4baac7 100644 --- a/manager/state/store/extensions.go +++ b/manager/state/store/extensions.go @@ -38,21 +38,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - extensions, err := FindExtensions(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Extensions)) + for i, x := range snapshot.Extensions { + toStoreObj[i] = extensionEntry{x} } - for _, e := range extensions { - if err := DeleteExtension(tx, e.ID); err != nil { - return err - } - } - for _, e := range snapshot.Extensions { - if err := CreateExtension(tx, e); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableExtension, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { @@ -80,6 +70,14 @@ func (e extensionEntry) CopyStoreObject() api.StoreObject { return extensionEntry{Extension: e.Extension.Copy()} } +// ensure that when update events are emitted, we unwrap extensionEntry +func (e extensionEntry) EventUpdate(oldObject api.StoreObject) api.Event { + if oldObject != nil { + return api.EventUpdateExtension{Extension: e.Extension, OldExtension: oldObject.(extensionEntry).Extension} + } + return api.EventUpdateExtension{Extension: e.Extension} +} + // CreateExtension adds a new extension to the store. // Returns ErrExist if the ID is already taken. func CreateExtension(tx Tx, e *api.Extension) error { diff --git a/manager/state/store/memory_test.go b/manager/state/store/memory_test.go index 3017b542ed..b4df214b0f 100644 --- a/manager/state/store/memory_test.go +++ b/manager/state/store/memory_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + events "github.com/docker/go-events" "github.com/docker/swarmkit/api" "github.com/docker/swarmkit/identity" "github.com/docker/swarmkit/manager/state" @@ -16,6 +17,43 @@ import ( ) var ( + clusterSet = []*api.Cluster{ + { + ID: "id1", + Spec: api.ClusterSpec{ + Annotations: api.Annotations{ + Name: "name1", + }, + }, + }, + { + ID: "id2", + Spec: api.ClusterSpec{ + Annotations: api.Annotations{ + Name: "name2", + }, + }, + }, + { + ID: "id3", + Spec: api.ClusterSpec{ + Annotations: api.Annotations{ + Name: "name3", + }, + }, + }, + } + altClusterSet = []*api.Cluster{ + { + ID: "alt-id1", + Spec: api.ClusterSpec{ + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + }, + } + nodeSet = []*api.Node{ { ID: "id1", @@ -49,6 +87,18 @@ var ( Role: api.NodeRoleWorker, }, } + altNodeSet = []*api.Node{ + { + ID: "alt-id1", + Spec: api.NodeSpec{ + Membership: api.NodeMembershipPending, + }, + Description: &api.NodeDescription{ + Hostname: "alt-name1", + }, + Role: api.NodeRoleManager, + }, + } serviceSet = []*api.Service{ { @@ -79,6 +129,16 @@ var ( }, }, } + altServiceSet = []*api.Service{ + { + ID: "alt-id1", + Spec: api.ServiceSpec{ + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + }, + } taskSet = []*api.Task{ { @@ -114,6 +174,19 @@ var ( DesiredState: api.TaskStateShutdown, }, } + altTaskSet = []*api.Task{ + { + ID: "alt-id1", + Annotations: api.Annotations{ + Name: "alt-name1", + }, + ServiceAnnotations: api.Annotations{ + Name: "alt-name1", + }, + DesiredState: api.TaskStateRunning, + NodeID: altNodeSet[0].ID, + }, + } networkSet = []*api.Network{ { @@ -141,28 +214,202 @@ var ( }, }, } + altNetworkSet = []*api.Network{ + { + ID: "alt-id1", + Spec: api.NetworkSpec{ + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + }, + } + + configSet = []*api.Config{ + { + ID: "id1", + Spec: api.ConfigSpec{ + Annotations: api.Annotations{ + Name: "name1", + }, + }, + }, + { + ID: "id2", + Spec: api.ConfigSpec{ + Annotations: api.Annotations{ + Name: "name2", + }, + }, + }, + { + ID: "id3", + Spec: api.ConfigSpec{ + Annotations: api.Annotations{ + Name: "name3", + }, + }, + }, + } + altConfigSet = []*api.Config{ + { + ID: "alt-id1", + Spec: api.ConfigSpec{ + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + }, + } + + secretSet = []*api.Secret{ + { + ID: "id1", + Spec: api.SecretSpec{ + Annotations: api.Annotations{ + Name: "name1", + }, + }, + }, + { + ID: "id2", + Spec: api.SecretSpec{ + Annotations: api.Annotations{ + Name: "name2", + }, + }, + }, + { + ID: "id3", + Spec: api.SecretSpec{ + Annotations: api.Annotations{ + Name: "name3", + }, + }, + }, + } + altSecretSet = []*api.Secret{ + { + ID: "alt-id1", + Spec: api.SecretSpec{ + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + }, + } + + extensionSet = []*api.Extension{ + { + ID: "id1", + Annotations: api.Annotations{ + Name: "name1", + }, + }, + { + ID: "id2", + Annotations: api.Annotations{ + Name: "name2", + }, + }, + { + ID: "id3", + Annotations: api.Annotations{ + Name: "name3", + }, + }, + } + altExtensionSet = []*api.Extension{ + { + ID: "alt-id1", + Annotations: api.Annotations{ + Name: "alt-name1", + }, + }, + } + + resourceSet = []*api.Resource{ + { + ID: "id1", + Annotations: api.Annotations{ + Name: "name1", + }, + Kind: "name1", // corresponds to extension id1 + }, + { + ID: "id2", + Annotations: api.Annotations{ + Name: "name2", + }, + Kind: "name2", // corresponds to extension id2 + }, + { + ID: "id3", + Annotations: api.Annotations{ + Name: "name3", + }, + Kind: "name3", // corresponds to extension id3 + }, + } + altResourceSet = []*api.Resource{ + { + ID: "alt-id1", + Annotations: api.Annotations{ + Name: "alt-name1", + }, + Kind: "alt-name1", // corresponds to extension alt-id1 + }, + } ) func setupTestStore(t *testing.T, s *MemoryStore) { + populateTestStore(t, s, + clusterSet, nodeSet, serviceSet, taskSet, networkSet, configSet, secretSet, + extensionSet, resourceSet) +} + +func populateTestStore(t *testing.T, s *MemoryStore, + clusters []*api.Cluster, nodes []*api.Node, services []*api.Service, tasks []*api.Task, networks []*api.Network, + configs []*api.Config, secrets []*api.Secret, extensions []*api.Extension, resources []*api.Resource) { err := s.Update(func(tx Tx) error { + // Prepoulate clusters + for _, c := range clusters { + assert.NoError(t, CreateCluster(tx, c)) + } + // Prepoulate nodes - for _, n := range nodeSet { + for _, n := range nodes { assert.NoError(t, CreateNode(tx, n)) } // Prepopulate services - for _, s := range serviceSet { + for _, s := range services { assert.NoError(t, CreateService(tx, s)) } // Prepopulate tasks - for _, task := range taskSet { + for _, task := range tasks { assert.NoError(t, CreateTask(tx, task)) } // Prepopulate networks - for _, n := range networkSet { + for _, n := range networks { assert.NoError(t, CreateNetwork(tx, n)) } - + // Prepopulate configs + for _, c := range configs { + assert.NoError(t, CreateConfig(tx, c)) + } + // Prepopulate secrets + for _, s := range secrets { + assert.NoError(t, CreateSecret(tx, s)) + } + // Prepopulate extensions + for _, c := range extensions { + assert.NoError(t, CreateExtension(tx, c)) + } + // Prepopulate resources + for _, s := range resources { + assert.NoError(t, CreateResource(tx, s)) + } return nil }) assert.NoError(t, err) @@ -1145,11 +1392,59 @@ func TestStoreSaveRestore(t *testing.T) { s2 := NewMemoryStore(nil) assert.NotNil(t, s2) + // setup s2 with the first element of each of the object sets (which should be + // updated on restore), as well as one extraneous object (which should be deleted + // on restore). We also want to bump the version on all the ones that will be + // updated just to make sure that restoration works. + version := api.Version{Index: 100} + c := clusterSet[0].Copy() + c.Meta.Version = version + n := nodeSet[0].Copy() + n.Meta.Version = version + s := serviceSet[0].Copy() + s.Meta.Version = version + task := taskSet[0].Copy() + task.Meta.Version = version + nw := networkSet[0].Copy() + nw.Meta.Version = version + cf := configSet[0].Copy() + cf.Meta.Version = version + sk := secretSet[0].Copy() + sk.Meta.Version = version + ext := extensionSet[0].Copy() + ext.Meta.Version = version + r := resourceSet[0].Copy() + r.Meta.Version = version + populateTestStore(t, s2, + append(altClusterSet, c), + append(altNodeSet, n), + append(altServiceSet, s), + append(altTaskSet, task), + append(altNetworkSet, nw), + append(altConfigSet, cf), + append(altSecretSet, sk), + append(altExtensionSet, ext), + append(altResourceSet, r), + ) + + watcher, cancel, err := ViewAndWatch(s2, func(ReadTx) error { + return nil + }) + assert.NoError(t, err) + defer cancel() - err := s2.Restore(snapshot) + err = s2.Restore(snapshot) assert.NoError(t, err) + // s2 should end up looking just like s1 s2.View(func(tx ReadTx) { + allClusters, err := FindClusters(tx, All) + assert.NoError(t, err) + assert.Len(t, allClusters, len(clusterSet)) + for i := range allClusters { + assert.Equal(t, allClusters[i], clusterSet[i]) + } + allTasks, err := FindTasks(tx, All) assert.NoError(t, err) assert.Len(t, allTasks, len(taskSet)) @@ -1177,7 +1472,245 @@ func TestStoreSaveRestore(t *testing.T) { for i := range allServices { assert.Equal(t, allServices[i], serviceSet[i]) } + + allConfigs, err := FindConfigs(tx, All) + assert.NoError(t, err) + assert.Len(t, allConfigs, len(configSet)) + for i := range allConfigs { + assert.Equal(t, allConfigs[i], configSet[i]) + } + + allSecrets, err := FindSecrets(tx, All) + assert.NoError(t, err) + assert.Len(t, allSecrets, len(secretSet)) + for i := range allSecrets { + assert.Equal(t, allSecrets[i], secretSet[i]) + } + + allExtensions, err := FindExtensions(tx, All) + assert.NoError(t, err) + assert.Len(t, allExtensions, len(extensionSet)) + for i := range allExtensions { + assert.Equal(t, allExtensions[i], extensionSet[i]) + } + + allResources, err := FindResources(tx, All) + assert.NoError(t, err) + assert.Len(t, allResources, len(resourceSet)) + for i := range allResources { + assert.Equal(t, allResources[i], resourceSet[i]) + } }) + + timeout := time.After(time.Second) + + // make sure we have 1 update event, 2 create events, and 1 delete event for each + // object type + var ( + clusterUpdates, clusterCreates, clusterDeletes, + nodeUpdates, nodeCreates, nodeDeletes, + serviceUpdates, serviceCreates, serviceDeletes, + taskUpdates, taskCreates, taskDeletes, + networkUpdates, networkCreates, networkDeletes, + configUpdates, configCreates, configDeletes, + secretUpdates, secretCreates, secretDeletes, + extensionUpdates, extensionCreates, extensionDeletes, + resourceUpdates, resourceCreates, resourceDeletes []api.StoreObject + ) + +waitForAllEvents: + for { + var update events.Event + select { + case update = <-watcher: + case <-timeout: + assert.FailNow(t, "did not get all the events we were expecting after a snapshot was restored") + } + + switch e := update.(type) { + + case api.EventUpdateCluster: + clusterUpdates = append(clusterUpdates, e.Cluster) + case api.EventCreateCluster: + clusterCreates = append(clusterCreates, e.Cluster) + case api.EventDeleteCluster: + clusterDeletes = append(clusterDeletes, e.Cluster) + + case api.EventUpdateNode: + nodeUpdates = append(nodeUpdates, e.Node) + case api.EventCreateNode: + nodeCreates = append(nodeCreates, e.Node) + case api.EventDeleteNode: + nodeDeletes = append(nodeDeletes, e.Node) + + case api.EventUpdateService: + serviceUpdates = append(serviceUpdates, e.Service) + case api.EventCreateService: + serviceCreates = append(serviceCreates, e.Service) + case api.EventDeleteService: + serviceDeletes = append(serviceDeletes, e.Service) + + case api.EventUpdateTask: + taskUpdates = append(taskUpdates, e.Task) + case api.EventCreateTask: + taskCreates = append(taskCreates, e.Task) + case api.EventDeleteTask: + taskDeletes = append(taskDeletes, e.Task) + + case api.EventUpdateNetwork: + networkUpdates = append(networkUpdates, e.Network) + case api.EventCreateNetwork: + networkCreates = append(networkCreates, e.Network) + case api.EventDeleteNetwork: + networkDeletes = append(networkDeletes, e.Network) + + case api.EventUpdateConfig: + configUpdates = append(configUpdates, e.Config) + case api.EventCreateConfig: + configCreates = append(configCreates, e.Config) + case api.EventDeleteConfig: + configDeletes = append(configDeletes, e.Config) + + case api.EventUpdateSecret: + secretUpdates = append(secretUpdates, e.Secret) + case api.EventCreateSecret: + secretCreates = append(secretCreates, e.Secret) + case api.EventDeleteSecret: + secretDeletes = append(secretDeletes, e.Secret) + + case api.EventUpdateExtension: + extensionUpdates = append(extensionUpdates, e.Extension) + case api.EventCreateExtension: + extensionCreates = append(extensionCreates, e.Extension) + case api.EventDeleteExtension: + extensionDeletes = append(extensionDeletes, e.Extension) + + case api.EventUpdateResource: + resourceUpdates = append(resourceUpdates, e.Resource) + case api.EventCreateResource: + resourceCreates = append(resourceCreates, e.Resource) + case api.EventDeleteResource: + resourceDeletes = append(resourceDeletes, e.Resource) + } + + // wait until we have all the events we want + for _, x := range [][]api.StoreObject{ + clusterUpdates, clusterDeletes, + nodeUpdates, nodeDeletes, + serviceUpdates, serviceDeletes, + taskUpdates, taskDeletes, + networkUpdates, networkDeletes, + configUpdates, configDeletes, + secretUpdates, secretDeletes, + extensionUpdates, extensionDeletes, + resourceUpdates, resourceDeletes, + } { + if len(x) < 1 { + continue waitForAllEvents + } + } + + for _, x := range [][]api.StoreObject{ + clusterCreates, + nodeCreates, + serviceCreates, + taskCreates, + networkCreates, + configCreates, + secretCreates, + extensionCreates, + resourceCreates, + } { + if len(x) < 2 { + continue waitForAllEvents + } + } + break + } + + assertHasSameIDs := func(changes []api.StoreObject, expected ...api.StoreObject) { + assert.Equal(t, len(expected), len(changes)) + expectedIDs := make(map[string]struct{}) + for _, s := range expected { + expectedIDs[s.GetID()] = struct{}{} + } + for _, s := range changes { + _, ok := expectedIDs[s.GetID()] + assert.True(t, ok) + } + } + + assertHasSameIDs(clusterUpdates, clusterSet[0]) + assertHasSameIDs(clusterDeletes, altClusterSet[0]) + cantCastArrays := make([]api.StoreObject, len(clusterSet[1:])) + for i, x := range clusterSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(clusterCreates, cantCastArrays...) + + assertHasSameIDs(nodeUpdates, nodeSet[0]) + assertHasSameIDs(nodeDeletes, altNodeSet[0]) + cantCastArrays = make([]api.StoreObject, len(nodeSet[1:])) + for i, x := range nodeSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(nodeCreates, cantCastArrays...) + + assertHasSameIDs(serviceUpdates, serviceSet[0]) + assertHasSameIDs(serviceDeletes, altServiceSet[0]) + cantCastArrays = make([]api.StoreObject, len(serviceSet[1:])) + for i, x := range serviceSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(serviceCreates, cantCastArrays...) + + assertHasSameIDs(taskUpdates, taskSet[0]) + assertHasSameIDs(taskDeletes, altTaskSet[0]) + cantCastArrays = make([]api.StoreObject, len(taskSet[1:])) + for i, x := range taskSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(taskCreates, cantCastArrays...) + + assertHasSameIDs(networkUpdates, networkSet[0]) + assertHasSameIDs(networkDeletes, altNetworkSet[0]) + cantCastArrays = make([]api.StoreObject, len(networkSet[1:])) + for i, x := range networkSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(networkCreates, cantCastArrays...) + + assertHasSameIDs(configUpdates, configSet[0]) + assertHasSameIDs(configDeletes, altConfigSet[0]) + cantCastArrays = make([]api.StoreObject, len(configSet[1:])) + for i, x := range configSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(configCreates, cantCastArrays...) + + assertHasSameIDs(secretUpdates, secretSet[0]) + assertHasSameIDs(secretDeletes, altSecretSet[0]) + cantCastArrays = make([]api.StoreObject, len(secretSet[1:])) + for i, x := range secretSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(secretCreates, cantCastArrays...) + + assertHasSameIDs(extensionUpdates, extensionSet[0]) + assertHasSameIDs(extensionDeletes, altExtensionSet[0]) + cantCastArrays = make([]api.StoreObject, len(extensionSet[1:])) + for i, x := range extensionSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(extensionCreates, cantCastArrays...) + + assertHasSameIDs(resourceUpdates, resourceSet[0]) + assertHasSameIDs(resourceDeletes, altResourceSet[0]) + cantCastArrays = make([]api.StoreObject, len(resourceSet[1:])) + for i, x := range resourceSet[1:] { + cantCastArrays[i] = x + } + assertHasSameIDs(resourceCreates, cantCastArrays...) } func TestWatchFrom(t *testing.T) { diff --git a/manager/state/store/networks.go b/manager/state/store/networks.go index df549d429c..3042def1bf 100644 --- a/manager/state/store/networks.go +++ b/manager/state/store/networks.go @@ -37,21 +37,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - networks, err := FindNetworks(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Networks)) + for i, x := range snapshot.Networks { + toStoreObj[i] = x } - for _, n := range networks { - if err := DeleteNetwork(tx, n.ID); err != nil { - return err - } - } - for _, n := range snapshot.Networks { - if err := CreateNetwork(tx, n); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableNetwork, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/nodes.go b/manager/state/store/nodes.go index 0661a16053..fa6ae85bd7 100644 --- a/manager/state/store/nodes.go +++ b/manager/state/store/nodes.go @@ -47,21 +47,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - nodes, err := FindNodes(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Nodes)) + for i, x := range snapshot.Nodes { + toStoreObj[i] = x } - for _, n := range nodes { - if err := DeleteNode(tx, n.ID); err != nil { - return err - } - } - for _, n := range snapshot.Nodes { - if err := CreateNode(tx, n); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableNode, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/object.go b/manager/state/store/object.go index 5f495fdfab..89029afb9f 100644 --- a/manager/state/store/object.go +++ b/manager/state/store/object.go @@ -13,3 +13,46 @@ type ObjectStoreConfig struct { Restore func(Tx, *api.StoreSnapshot) error ApplyStoreAction func(Tx, api.StoreAction) error } + +// RestoreTable takes a list of new objects of a particular type (e.g. clusters, +// nodes, etc., which conform to the StoreObject interface) and replaces the +// existing objects in the store of that type with the new objects. +func RestoreTable(tx Tx, table string, newObjects []api.StoreObject) error { + checkType := func(by By) error { + return nil + } + var oldObjects []api.StoreObject + appendResult := func(o api.StoreObject) { + oldObjects = append(oldObjects, o) + } + + err := tx.find(table, All, checkType, appendResult) + if err != nil { + return nil + } + + updated := make(map[string]struct{}) + + for _, o := range newObjects { + objectID := o.GetID() + if existing := tx.lookup(table, indexID, objectID); existing != nil { + if err := tx.update(table, o); err != nil { + return err + } + updated[objectID] = struct{}{} + } else { + if err := tx.create(table, o); err != nil { + return err + } + } + } + for _, o := range oldObjects { + objectID := o.GetID() + if _, ok := updated[objectID]; !ok { + if err := tx.delete(table, objectID); err != nil { + return err + } + } + } + return nil +} diff --git a/manager/state/store/resources.go b/manager/state/store/resources.go index 4e1ec7154b..1f2c3904f1 100644 --- a/manager/state/store/resources.go +++ b/manager/state/store/resources.go @@ -40,21 +40,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - resources, err := FindResources(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Resources)) + for i, x := range snapshot.Resources { + toStoreObj[i] = resourceEntry{x} } - for _, r := range resources { - if err := DeleteResource(tx, r.ID); err != nil { - return err - } - } - for _, r := range snapshot.Resources { - if err := CreateResource(tx, r); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableResource, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { @@ -82,6 +72,14 @@ func (r resourceEntry) CopyStoreObject() api.StoreObject { return resourceEntry{Resource: r.Resource.Copy()} } +// ensure that when update events are emitted, we unwrap resourceEntry +func (r resourceEntry) EventUpdate(oldObject api.StoreObject) api.Event { + if oldObject != nil { + return api.EventUpdateResource{Resource: r.Resource, OldResource: oldObject.(resourceEntry).Resource} + } + return api.EventUpdateResource{Resource: r.Resource} +} + func confirmExtension(tx Tx, r *api.Resource) error { // There must be an extension corresponding to the Kind field. extensions, err := FindExtensions(tx, ByName(r.Kind)) diff --git a/manager/state/store/secrets.go b/manager/state/store/secrets.go index 203c94c974..bf5653fd72 100644 --- a/manager/state/store/secrets.go +++ b/manager/state/store/secrets.go @@ -37,21 +37,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - secrets, err := FindSecrets(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Secrets)) + for i, x := range snapshot.Secrets { + toStoreObj[i] = x } - for _, s := range secrets { - if err := DeleteSecret(tx, s.ID); err != nil { - return err - } - } - for _, s := range snapshot.Secrets { - if err := CreateSecret(tx, s); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableSecret, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/services.go b/manager/state/store/services.go index 884bc3533d..1adbb87fe4 100644 --- a/manager/state/store/services.go +++ b/manager/state/store/services.go @@ -58,21 +58,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - services, err := FindServices(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Services)) + for i, x := range snapshot.Services { + toStoreObj[i] = x } - for _, s := range services { - if err := DeleteService(tx, s.ID); err != nil { - return err - } - } - for _, s := range snapshot.Services { - if err := CreateService(tx, s); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableService, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) { diff --git a/manager/state/store/tasks.go b/manager/state/store/tasks.go index b8d8a731c9..bf31d764fb 100644 --- a/manager/state/store/tasks.go +++ b/manager/state/store/tasks.go @@ -82,21 +82,11 @@ func init() { return err }, Restore: func(tx Tx, snapshot *api.StoreSnapshot) error { - tasks, err := FindTasks(tx, All) - if err != nil { - return err + toStoreObj := make([]api.StoreObject, len(snapshot.Tasks)) + for i, x := range snapshot.Tasks { + toStoreObj[i] = x } - for _, t := range tasks { - if err := DeleteTask(tx, t.ID); err != nil { - return err - } - } - for _, t := range snapshot.Tasks { - if err := CreateTask(tx, t); err != nil { - return err - } - } - return nil + return RestoreTable(tx, tableTask, toStoreObj) }, ApplyStoreAction: func(tx Tx, sa api.StoreAction) error { switch v := sa.Target.(type) {