From 8110e0eab3399227f43cdf60bf527f69ad1a7e00 Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Mon, 15 Nov 2021 12:22:39 -0500 Subject: [PATCH] Increment version of host during updates (#1707) Currently host does not use a version, which means it's not protected against race conditions of either itself or its value objects. This adds and plumbs through version to ensure that if a host or its value objects change that they are not going to be affected by a race condition. --- .../postgres/20/06_plugin_host_views.up.sql | 5 +-- .../postgres/20/08_plugin_host_views.up.sql | 41 +++++++++++++++++ internal/host/plugin/host.go | 2 + internal/host/plugin/job_set_sync.go | 44 ++++++++++++------- internal/host/plugin/job_set_sync_test.go | 22 +++++++++- internal/host/plugin/repository_host_test.go | 1 + internal/host/plugin/repository_host_util.go | 4 ++ internal/host/plugin/store/host.pb.go | 5 ++- internal/host/plugin/testing.go | 1 + .../storage/host/plugin/store/v1/host.proto | 3 +- .../controller/handlers/hosts/host_service.go | 4 -- .../handlers/hosts/host_service_test.go | 2 +- 12 files changed, 106 insertions(+), 28 deletions(-) create mode 100644 internal/db/schema/migrations/oss/postgres/20/08_plugin_host_views.up.sql diff --git a/internal/db/schema/migrations/oss/postgres/20/06_plugin_host_views.up.sql b/internal/db/schema/migrations/oss/postgres/20/06_plugin_host_views.up.sql index f3e476457..95f76646e 100644 --- a/internal/db/schema/migrations/oss/postgres/20/06_plugin_host_views.up.sql +++ b/internal/db/schema/migrations/oss/postgres/20/06_plugin_host_views.up.sql @@ -53,10 +53,7 @@ from comment on view host_plugin_catalog_with_secret is 'host plugin catalog with its associated persisted data'; --- host_plugin_host_with_value_obj_and_set_memberships is useful for reading a --- plugin host with its associated value objects (ip addresses, dns names) and --- set membership as columns with delimited values. The delimiter depends on --- the value objects (e.g. if they need ordering). +-- REPLACED in 20/08_plugin_host_views.up.sql create view host_plugin_host_with_value_obj_and_set_memberships as select h.public_id, diff --git a/internal/db/schema/migrations/oss/postgres/20/08_plugin_host_views.up.sql b/internal/db/schema/migrations/oss/postgres/20/08_plugin_host_views.up.sql new file mode 100644 index 000000000..52e4ba9e1 --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/20/08_plugin_host_views.up.sql @@ -0,0 +1,41 @@ +begin; + +alter table host_plugin_host + add column version wt_version; + +create trigger update_version_column after update on host_plugin_host + for each row execute procedure update_version_column(); + +drop view host_plugin_host_with_value_obj_and_set_memberships; + +-- host_plugin_host_with_value_obj_and_set_memberships is useful for reading a +-- plugin host with its associated value objects (ip addresses, dns names) and +-- set membership as columns with delimited values. The delimiter depends on +-- the value objects (e.g. if they need ordering). +create view host_plugin_host_with_value_obj_and_set_memberships as +select + h.public_id, + h.catalog_id, + h.external_id, + hc.scope_id, + hc.plugin_id, + h.name, + h.description, + h.create_time, + h.update_time, + h.version, + -- the string_agg(..) column will be null if there are no associated value objects + string_agg(distinct host(hip.address), '|') as ip_addresses, + string_agg(distinct hdns.name, '|') as dns_names, + string_agg(distinct hpsm.set_id, '|') as set_ids +from + host_plugin_host h + join host_plugin_catalog hc on h.catalog_id = hc.public_id + left outer join host_ip_address hip on h.public_id = hip.host_id + left outer join host_dns_name hdns on h.public_id = hdns.host_id + left outer join host_plugin_set_member hpsm on h.public_id = hpsm.host_id +group by h.public_id, hc.plugin_id, hc.scope_id; +comment on view host_plugin_host_with_value_obj_and_set_memberships is + 'host plugin host with its associated value objects'; + +commit; \ No newline at end of file diff --git a/internal/host/plugin/host.go b/internal/host/plugin/host.go index b83e9dd85..93c7671e2 100644 --- a/internal/host/plugin/host.go +++ b/internal/host/plugin/host.go @@ -116,6 +116,7 @@ type hostAgg struct { Description string CreateTime *timestamp.Timestamp UpdateTime *timestamp.Timestamp + Version uint32 IpAddresses string DnsNames string SetIds string @@ -132,6 +133,7 @@ func (agg *hostAgg) toHost() *Host { h.Description = agg.Description h.CreateTime = agg.CreateTime h.UpdateTime = agg.UpdateTime + h.Version = agg.Version if agg.IpAddresses != "" { h.IpAddresses = strings.Split(agg.IpAddresses, aggregateDelimiter) diff --git a/internal/host/plugin/job_set_sync.go b/internal/host/plugin/job_set_sync.go index e2727f395..f1d45f293 100644 --- a/internal/host/plugin/job_set_sync.go +++ b/internal/host/plugin/job_set_sync.go @@ -414,22 +414,36 @@ func (r *SetSyncJob) upsertHosts( db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { msgs := make([]*oplog.Message, 0) - ticket, err := w.GetTicket(hi.h) + ticket, err := w.GetTicket(ret) if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get ticket")) } - if hi.dirtyHost { - var hOplogMsg oplog.Message - onConflict := &db.OnConflict{ - Target: db.Constraint("host_plugin_host_pkey"), - Action: db.SetColumns([]string{"name", "description"}), - } - if err := w.Create(ctx, ret, db.NewOplogMsg(&hOplogMsg), db.WithOnConflict(onConflict)); err != nil { - return errors.Wrap(ctx, err, op) - } - msgs = append(msgs, &hOplogMsg) + // We always write the host itself as we need to update version + // for optimistic locking for any value object update. + var hOplogMsg oplog.Message + onConflict := &db.OnConflict{ + Target: db.Constraint("host_plugin_host_pkey"), + Action: db.SetColumns([]string{"name", "description", "version"}), } + var rowsAffected int64 + dbOpts := []db.Option{ + db.NewOplogMsg(&hOplogMsg), + db.WithOnConflict(onConflict), + db.WithReturnRowsAffected(&rowsAffected), + } + version := ret.Version + if version > 0 { + dbOpts = append(dbOpts, db.WithVersion(&version)) + ret.Version += 1 + } + if err := w.Create(ctx, ret, dbOpts...); err != nil { + return errors.Wrap(ctx, err, op) + } + if rowsAffected != 1 { + return errors.New(ctx, errors.UnexpectedRowsAffected, op, "no rows affected during upsert") + } + msgs = append(msgs, &hOplogMsg) // IP handling { @@ -440,7 +454,7 @@ func (r *SetSyncJob) upsertHosts( return err } if count != len(hi.ipsToRemove) { - return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d ips from host %s, removed %d", len(hi.ipsToRemove), hi.h.PublicId, count)) + return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d ips from host %s, removed %d", len(hi.ipsToRemove), ret.PublicId, count)) } msgs = append(msgs, oplogMsgs...) } @@ -451,7 +465,7 @@ func (r *SetSyncJob) upsertHosts( Action: db.DoNothing(true), } if err := w.CreateItems(ctx, hi.ipsToAdd.toArray(), db.NewOplogMsgs(&oplogMsgs), db.WithOnConflict(onConflict)); err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("adding ips %v for host %q", hi.ipsToAdd.toArray(), hi.h.GetPublicId()))) + return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("adding ips %v for host %q", hi.ipsToAdd.toArray(), ret.GetPublicId()))) } msgs = append(msgs, oplogMsgs...) } @@ -466,7 +480,7 @@ func (r *SetSyncJob) upsertHosts( return err } if count != len(hi.dnsNamesToRemove) { - return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d dns names from host %s, removed %d", len(hi.dnsNamesToRemove), hi.h.PublicId, count)) + return errors.New(ctx, errors.UnexpectedRowsAffected, op, fmt.Sprintf("expected to remove %d dns names from host %s, removed %d", len(hi.dnsNamesToRemove), ret.PublicId, count)) } msgs = append(msgs, oplogMsgs...) } @@ -483,7 +497,7 @@ func (r *SetSyncJob) upsertHosts( } } - metadata := hi.h.oplog(oplog.OpType_OP_TYPE_UPDATE) + metadata := ret.oplog(oplog.OpType_OP_TYPE_UPDATE) if err := w.WriteOplogEntryWith(ctx, oplogWrapper, ticket, metadata, msgs); err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write oplog")) } diff --git a/internal/host/plugin/job_set_sync_test.go b/internal/host/plugin/job_set_sync_test.go index 63c3c6c94..e7432b3ef 100644 --- a/internal/host/plugin/job_set_sync_test.go +++ b/internal/host/plugin/job_set_sync_test.go @@ -2,6 +2,7 @@ package plugin import ( "context" + "fmt" "testing" "time" @@ -159,17 +160,19 @@ func TestSetSyncJob_Run(t *testing.T) { cat := TestCatalog(t, conn, prj.GetPublicId(), plg.GetPublicId()) set1 := TestSet(t, conn, kmsCache, sched, cat, plgm) + counter := new(uint32) plgServer.ListHostsFn = func(ctx context.Context, req *plgpb.ListHostsRequest) (*plgpb.ListHostsResponse, error) { assert.GreaterOrEqual(1, len(req.GetSets())) var setIds []string for _, s := range req.GetSets() { setIds = append(setIds, s.GetId()) } + *counter += 1 return &plgpb.ListHostsResponse{ Hosts: []*plgpb.ListHostsResponseHost{ { ExternalId: "first", - IpAddresses: []string{"10.0.0.1"}, + IpAddresses: []string{fmt.Sprintf("10.0.0.%d", *counter)}, DnsNames: []string{"foo.com"}, SetIds: setIds, }, @@ -177,6 +180,9 @@ func TestSetSyncJob_Run(t *testing.T) { }, nil } + hostRepo, err := NewRepository(rw, rw, kmsCache, sche, plgm) + require.NoError(err) + hsa := &hostSetAgg{PublicId: set1.GetPublicId()} require.NoError(rw.LookupByPublicId(ctx, hsa)) assert.Less(hsa.LastSyncTime.AsTime().UnixNano(), hsa.CreateTime.AsTime().UnixNano()) @@ -187,6 +193,13 @@ func TestSetSyncJob_Run(t *testing.T) { // The single existing set should have been processed assert.Equal(1, r.numSets) assert.Equal(1, r.numProcessed) + // Check the version number of the host(s) + hosts, _, err := hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId) + require.NoError(err) + assert.Len(hosts, 1) + for _, host := range hosts { + assert.Equal(uint32(1), host.Version) + } require.NoError(rw.LookupByPublicId(ctx, hsa)) assert.Greater(hsa.LastSyncTime.AsTime().UnixNano(), hsa.CreateTime.AsTime().UnixNano()) @@ -215,6 +228,13 @@ func TestSetSyncJob_Run(t *testing.T) { // The single existing set should have been processed assert.Equal(1, r.numSets) assert.Equal(1, r.numProcessed) + // Check the version number of the host(s) again + hosts, _, err = hostRepo.ListHostsByCatalogId(ctx, hsa.CatalogId) + require.NoError(err) + assert.Len(hosts, 1) + for _, host := range hosts { + assert.Equal(uint32(2), host.Version) + } // Run sync with a new second set _ = TestSet(t, conn, kmsCache, sched, cat, plgm) diff --git a/internal/host/plugin/repository_host_test.go b/internal/host/plugin/repository_host_test.go index 002d2f660..0bf601f6e 100644 --- a/internal/host/plugin/repository_host_test.go +++ b/internal/host/plugin/repository_host_test.go @@ -175,6 +175,7 @@ func TestJob_UpsertHosts(t *testing.T) { ph.SetIds = ph.SetIds[0 : len(ph.SetIds)-1] e.SetIds = e.SetIds[0 : len(e.SetIds)-1] + e.Version++ return &input{ catalog: catalog, sets: setIds, diff --git a/internal/host/plugin/repository_host_util.go b/internal/host/plugin/repository_host_util.go index fb8cd97db..28f023e23 100644 --- a/internal/host/plugin/repository_host_util.go +++ b/internal/host/plugin/repository_host_util.go @@ -75,6 +75,10 @@ func createNewHostMap(ctx context.Context, // assume catalog ID and external ID don't change because if they do // the public ID will be different as well. currHost := currentHostMap[newHost.PublicId] + if currHost != nil { + hi.h.Version = currHost.Version + } + switch { case currHost == nil, currHost.Name != newHost.Name, diff --git a/internal/host/plugin/store/host.pb.go b/internal/host/plugin/store/host.pb.go index 0fa003b26..df65c552f 100644 --- a/internal/host/plugin/store/host.pb.go +++ b/internal/host/plugin/store/host.pb.go @@ -447,8 +447,9 @@ type Host struct { // plugin_host_catalog and must be set. // @inject_tag: `gorm:"not_null"` CatalogId string `protobuf:"bytes,7,opt,name=catalog_id,json=catalogId,proto3" json:"catalog_id,omitempty" gorm:"not_null"` - // @inject_tag: `gorm:"-"` - Version uint32 `protobuf:"varint,8,opt,name=version,proto3" json:"version,omitempty" gorm:"-"` + // version allows optimistic locking of the resource + // @inject_tag: `gorm:"default:null"` + Version uint32 `protobuf:"varint,8,opt,name=version,proto3" json:"version,omitempty" gorm:"default:null"` // ip_addresses are the ip addresses associated with this host and will // be persisted in the db through the HostAddress message. // @inject_tag: `gorm:"-"` diff --git a/internal/host/plugin/testing.go b/internal/host/plugin/testing.go index 6f6b2a62f..d15df853f 100644 --- a/internal/host/plugin/testing.go +++ b/internal/host/plugin/testing.go @@ -189,6 +189,7 @@ func TestExternalHosts(t *testing.T, catalog *HostCatalog, setIds []string, coun ExternalId: externalId, IpAddresses: []string{ipStr}, DnsNames: []string{dnsName}, + Version: 1, }, }) } diff --git a/internal/proto/local/controller/storage/host/plugin/store/v1/host.proto b/internal/proto/local/controller/storage/host/plugin/store/v1/host.proto index fa40064aa..b965ab49e 100644 --- a/internal/proto/local/controller/storage/host/plugin/store/v1/host.proto +++ b/internal/proto/local/controller/storage/host/plugin/store/v1/host.proto @@ -160,7 +160,8 @@ message Host { // @inject_tag: `gorm:"not_null"` string catalog_id = 7; - // @inject_tag: `gorm:"-"` + // version allows optimistic locking of the resource + // @inject_tag: `gorm:"default:null"` uint32 version = 8; // ip_addresses are the ip addresses associated with this host and will diff --git a/internal/servers/controller/handlers/hosts/host_service.go b/internal/servers/controller/handlers/hosts/host_service.go index 69548ba85..1e59df85b 100644 --- a/internal/servers/controller/handlers/hosts/host_service.go +++ b/internal/servers/controller/handlers/hosts/host_service.go @@ -598,10 +598,6 @@ func toProto(ctx context.Context, in host.Host, opt ...handlers.Option) (*pb.Hos if outputFields.Has(globals.ExternalIdField) { out.ExternalId = h.ExternalId } - // FIXME? Seems we should be _using_ a version... - if outputFields.Has(globals.VersionField) { - out.Version = 1 - } } return &out, nil } diff --git a/internal/servers/controller/handlers/hosts/host_service_test.go b/internal/servers/controller/handlers/hosts/host_service_test.go index 946651663..c5d43d2c7 100644 --- a/internal/servers/controller/handlers/hosts/host_service_test.go +++ b/internal/servers/controller/handlers/hosts/host_service_test.go @@ -758,7 +758,7 @@ func TestCreate(t *testing.T) { } } -func TestUpdate(t *testing.T) { +func TestUpdate_Static(t *testing.T) { t.Parallel() conn, _ := db.TestSetup(t, "postgres") wrapper := db.TestWrapper(t)