From 1830605a7e3e47d77b4004c864cda7c3b5bef8db Mon Sep 17 00:00:00 2001 From: Haotian Date: Fri, 3 Jun 2022 11:59:02 -0700 Subject: [PATCH] refact(servers) Update api for new worker and controller server tables (#2101) * Discontinue support for old Server message in ServerCoordinationService. Split message into ServerWorkerStatus and UpstreamServer (which currently contains controller addresses). Remove unnecessary fields. Modify workerStatusServer to use new fields. Co-authored-by: Todd --- .../daemon/cluster/handlers/worker_service.go | 55 ++- .../handlers/worker_service_status_test.go | 103 +++-- .../targets/tcp/target_service_test.go | 31 +- internal/daemon/worker/status.go | 18 +- internal/daemon/worker/testing.go | 2 +- internal/daemon/worker/worker.go | 13 +- .../daemon/worker/worker_proxy_service.go | 2 +- .../server_coordination_service.pb.go | 414 ++++++++++++------ .../v1/server_coordination_service.proto | 39 +- .../proto/controller/servers/v1/servers.proto | 47 +- internal/servers/options.go | 8 + internal/servers/options_test.go | 8 + internal/servers/repository_worker.go | 13 +- internal/servers/servers.pb.go | 234 ++++------ internal/tests/helper/testing_helper.go | 2 +- 15 files changed, 566 insertions(+), 423 deletions(-) diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 1ddda90c5..ac7e03aee 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -58,30 +58,42 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques const op = "workers.(workerServiceServer).Status" // TODO: on the worker, if we get errors back from this repeatedly, do we // terminate all sessions since we can't know if they were canceled? - ws.updateTimes.Store(req.Worker.PrivateId, time.Now()) + + wStat := req.GetWorkerStatus() + switch { + case wStat.GetName() == "": + return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Name is not set in the request but is required.") + case wStat.GetAddress() == "": + return &pbs.StatusResponse{}, status.Error(codes.InvalidArgument, "Address is not set but is required.") + } + // This Store call is currently only for testing purposes + ws.updateTimes.Store(wStat.GetName(), time.Now()) + serverRepo, err := ws.serversRepoFn() if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error getting servers repo")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error acquiring repo to store worker status: %v", err) } - reqServer := req.GetWorker() // Convert API tags to storage tags - workerTags := make([]*servers.Tag, 0, len(reqServer.Tags)) - for k, v := range reqServer.Tags { - for _, val := range v.GetValues() { - workerTags = append(workerTags, &servers.Tag{ - Key: k, - Value: val, - }) - } + wTags := wStat.GetTags() + workerTags := make([]*servers.Tag, 0, len(wTags)) + for _, v := range wTags { + workerTags = append(workerTags, &servers.Tag{ + Key: v.GetKey(), + Value: v.GetValue(), + }) } wConf := servers.NewWorkerForStatus(scope.Global.String(), - servers.WithName(reqServer.PrivateId), - servers.WithAddress(reqServer.Address), + servers.WithName(wStat.GetName()), + servers.WithAddress(wStat.GetAddress()), servers.WithWorkerTags(workerTags...)) - wrk, err := serverRepo.UpsertWorkerStatus(ctx, wConf, servers.WithUpdateTags(req.GetUpdateTags())) + opts := []servers.Option{servers.WithUpdateTags(req.GetUpdateTags())} + if wStat.GetPublicId() != "" { + opts = append(opts, servers.WithPublicId(wStat.GetPublicId())) + } + wrk, err := serverRepo.UpsertWorkerStatus(ctx, wConf, opts...) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error storing worker status")) return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error storing worker status: %v", err) @@ -92,18 +104,17 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques return &pbs.StatusResponse{}, status.Errorf(codes.Internal, "Error getting current controllers: %v", err) } - responseControllers := []*servers.Server{} + responseControllers := []*pbs.UpstreamServer{} for _, c := range controllers { - thisController := &servers.Server{ - PrivateId: c.PrivateId, - Address: c.Address, - CreateTime: c.CreateTime, - UpdateTime: c.UpdateTime, + thisController := &pbs.UpstreamServer{ + Address: c.Address, + Type: pbs.UpstreamServer_TYPE_CONTROLLER, } responseControllers = append(responseControllers, thisController) } ret := &pbs.StatusResponse{ - Controllers: responseControllers, + CalculatedUpstreams: responseControllers, + WorkerId: wrk.GetPublicId(), } stateReport := make([]session.StateReport, 0, len(req.GetJobs())) @@ -152,8 +163,8 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques notActive, err := session.WorkerStatusReport(ctx, sessRepo, connectionRepo, wrk.GetPublicId(), stateReport) if err != nil { return nil, status.Errorf(codes.Internal, - "Error comparing state of sessions for worker %q with public id %q: %v", - req.Worker.PrivateId, wrk.GetPublicId(), err) + "Error comparing state of sessions for worker with public id %q: %v", + wrk.GetPublicId(), err) } for _, na := range notActive { var connChanges []*pbs.Connection diff --git a/internal/daemon/cluster/handlers/worker_service_status_test.go b/internal/daemon/cluster/handlers/worker_service_status_test.go index 67112c63d..d34924890 100644 --- a/internal/daemon/cluster/handlers/worker_service_status_test.go +++ b/internal/daemon/cluster/handlers/worker_service_status_test.go @@ -21,6 +21,8 @@ import ( "github.com/hashicorp/boundary/internal/target/tcp" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestStatus(t *testing.T) { @@ -98,31 +100,30 @@ func TestStatus(t *testing.T) { name: "No Sessions", wantErr: false, req: &pbs.StatusRequest{ - Worker: &servers.Server{ - PrivateId: worker1.GetWorkerReportedName(), - Address: worker1.CanonicalAddress(), - CreateTime: worker1.CreateTime, - UpdateTime: worker1.UpdateTime, + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetWorkerReportedName(), + Address: worker1.CanonicalAddress(), }, }, want: &pbs.StatusResponse{ - Controllers: []*servers.Server{ + CalculatedUpstreams: []*pbs.UpstreamServer{ { - PrivateId: "test_controller1", - Address: "127.0.0.1", + Type: pbs.UpstreamServer_TYPE_CONTROLLER, + Address: "127.0.0.1", }, }, + WorkerId: worker1.PublicId, }, }, { name: "Still Active", wantErr: false, req: &pbs.StatusRequest{ - Worker: &servers.Server{ - PrivateId: worker1.GetWorkerReportedName(), - Address: worker1.CanonicalAddress(), - CreateTime: worker1.CreateTime, - UpdateTime: worker1.UpdateTime, + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetWorkerReportedName(), + Address: worker1.CanonicalAddress(), }, Jobs: []*pbs.JobStatus{ { @@ -145,14 +146,37 @@ func TestStatus(t *testing.T) { }, }, want: &pbs.StatusResponse{ - Controllers: []*servers.Server{ + CalculatedUpstreams: []*pbs.UpstreamServer{ { - PrivateId: "test_controller1", - Address: "127.0.0.1", + Type: pbs.UpstreamServer_TYPE_CONTROLLER, + Address: "127.0.0.1", }, }, + WorkerId: worker1.PublicId, }, }, + { + name: "No Name", + wantErr: true, + req: &pbs.StatusRequest{ + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Address: worker1.CanonicalAddress(), + }, + }, + wantErrMsg: status.Error(codes.InvalidArgument, "Name is not set in the request but is required.").Error(), + }, + { + name: "No Address", + wantErr: true, + req: &pbs.StatusRequest{ + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetWorkerReportedName(), + }, + }, + wantErrMsg: status.Error(codes.InvalidArgument, "Address is not set but is required.").Error(), + }, } for _, tc := range cases { @@ -162,7 +186,7 @@ func TestStatus(t *testing.T) { got, err := s.Status(ctx, tc.req) if tc.wantErr { require.Error(err) - assert.Nil(got) + assert.Equal(got, &pbs.StatusResponse{}) assert.Equal(tc.wantErrMsg, err.Error()) return } @@ -172,14 +196,15 @@ func TestStatus(t *testing.T) { got, cmpopts.IgnoreUnexported( pbs.StatusResponse{}, - servers.Server{}, + servers.ServerWorkerStatus{}, + pbs.UpstreamServer{}, pbs.JobChangeRequest{}, pbs.Job{}, pbs.Job_SessionInfo{}, pbs.SessionJobInfo{}, pbs.Connection{}, ), - cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"), + cmpopts.IgnoreFields(servers.ServerWorkerStatus{}, "Tags"), ), ) }) @@ -278,11 +303,10 @@ func TestStatusSessionClosed(t *testing.T) { require.NoError(t, err) }, req: &pbs.StatusRequest{ - Worker: &servers.Server{ - PrivateId: worker1.GetWorkerReportedName(), - Address: worker1.CanonicalAddress(), - CreateTime: worker1.CreateTime, - UpdateTime: worker1.UpdateTime, + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetWorkerReportedName(), + Address: worker1.CanonicalAddress(), }, Jobs: []*pbs.JobStatus{ { @@ -305,10 +329,10 @@ func TestStatusSessionClosed(t *testing.T) { }, }, want: &pbs.StatusResponse{ - Controllers: []*servers.Server{ + CalculatedUpstreams: []*pbs.UpstreamServer{ { - PrivateId: "test_controller1", - Address: "127.0.0.1", + Type: pbs.UpstreamServer_TYPE_CONTROLLER, + Address: "127.0.0.1", }, }, JobsRequests: []*pbs.JobChangeRequest{ @@ -325,6 +349,7 @@ func TestStatusSessionClosed(t *testing.T) { RequestType: pbs.CHANGETYPE_CHANGETYPE_UPDATE_STATE, }, }, + WorkerId: worker1.PublicId, }, }, } @@ -349,14 +374,15 @@ func TestStatusSessionClosed(t *testing.T) { got, cmpopts.IgnoreUnexported( pbs.StatusResponse{}, - servers.Server{}, + servers.ServerWorkerStatus{}, + pbs.UpstreamServer{}, pbs.JobChangeRequest{}, pbs.Job{}, pbs.Job_SessionInfo{}, pbs.SessionJobInfo{}, pbs.Connection{}, ), - cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"), + cmpopts.IgnoreFields(servers.ServerWorkerStatus{}, "Tags"), ), ) }) @@ -444,9 +470,10 @@ func TestStatusDeadConnection(t *testing.T) { require.NotEqual(t, deadConn.PublicId, connection.PublicId) req := &pbs.StatusRequest{ - Worker: &servers.Server{ - PrivateId: worker1.GetWorkerReportedName(), - Address: worker1.GetWorkerReportedAddress(), + WorkerStatus: &servers.ServerWorkerStatus{ + PublicId: worker1.GetPublicId(), + Name: worker1.GetWorkerReportedName(), + Address: worker1.CanonicalAddress(), }, Jobs: []*pbs.JobStatus{ { @@ -469,12 +496,13 @@ func TestStatusDeadConnection(t *testing.T) { }, } want := &pbs.StatusResponse{ - Controllers: []*servers.Server{ + CalculatedUpstreams: []*pbs.UpstreamServer{ { - PrivateId: "test_controller1", - Address: "127.0.0.1", + Type: pbs.UpstreamServer_TYPE_CONTROLLER, + Address: "127.0.0.1", }, }, + WorkerId: worker1.PublicId, } got, err := s.Status(ctx, req) @@ -484,14 +512,15 @@ func TestStatusDeadConnection(t *testing.T) { got, cmpopts.IgnoreUnexported( pbs.StatusResponse{}, - servers.Server{}, + servers.ServerWorkerStatus{}, + pbs.UpstreamServer{}, pbs.JobChangeRequest{}, pbs.Job{}, pbs.Job_SessionInfo{}, pbs.SessionJobInfo{}, pbs.Connection{}, ), - cmpopts.IgnoreFields(servers.Server{}, "CreateTime", "UpdateTime"), + cmpopts.IgnoreFields(servers.ServerWorkerStatus{}, "Tags"), ), ) diff --git a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go index 26120d855..940963743 100644 --- a/internal/daemon/controller/handlers/targets/tcp/target_service_test.go +++ b/internal/daemon/controller/handlers/targets/tcp/target_service_test.go @@ -2646,9 +2646,6 @@ func TestAuthorizeSession(t *testing.T) { sessionRepoFn := func() (*session.Repository, error) { return session.NewRepository(rw, rw, kms) } - connectionRepoFn := func() (*session.ConnectionRepository, error) { - return session.NewConnectionRepository(ctx, rw, rw, kms) - } staticHostRepoFn := func() (*static.Repository, error) { return static.NewRepository(rw, rw, kms) } @@ -2787,14 +2784,7 @@ func TestAuthorizeSession(t *testing.T) { require.NoError(t, err) // Tell our DB that there is a worker ready to serve the data - workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms) - _, err = workerService.Status(ctx, &spbs.StatusRequest{ - Worker: &spb.Server{ - PrivateId: "w_1234567890", - Address: "localhost:8457", - }, - }) - require.NoError(t, err) + servers.TestWorker(t, conn, wrapper) asRes1, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{ Id: tar.GetPublicId(), @@ -2903,9 +2893,6 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) { sessionRepoFn := func() (*session.Repository, error) { return session.NewRepository(rw, rw, kms) } - connectionRepoFn := func() (*session.ConnectionRepository, error) { - return session.NewConnectionRepository(ctx, rw, rw, kms) - } staticHostRepoFn := func() (*static.Repository, error) { return static.NewRepository(rw, rw, kms) } @@ -3075,14 +3062,7 @@ func TestAuthorizeSessionTypedCredentials(t *testing.T) { require.NoError(t, err) // Tell our DB that there is a worker ready to serve the data - workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms) - _, err = workerService.Status(ctx, &spbs.StatusRequest{ - Worker: &spb.Server{ - PrivateId: "w_1234567890", - Address: "localhost:8457", - }, - }) - require.NoError(t, err) + servers.TestWorker(t, conn, wrapper) asRes, err := s.AuthorizeSession(ctx, &pbs.AuthorizeSessionRequest{ Id: tar.GetPublicId(), @@ -3208,9 +3188,10 @@ func TestAuthorizeSession_Errors(t *testing.T) { workerExists := func(tar target.Target) (version uint32) { workerService := cluster.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, &sync.Map{}, kms) _, err := workerService.Status(context.Background(), &spbs.StatusRequest{ - Worker: &spb.Server{ - PrivateId: "w_1234567890", - Address: "localhost:123", + WorkerStatus: &spb.ServerWorkerStatus{ + PublicId: "w_1234567890", + Name: "w_1234567890", + Address: "localhost:123", }, }) require.NoError(t, err) diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index 4ddad3862..728e0fe63 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -135,21 +135,21 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) { // Send status information client := w.controllerStatusConn.Load().(pbs.ServerCoordinationServiceClient) - var tags map[string]*servers.TagValues + var tags []*servers.TagPair // If we're not going to request a tag update, no reason to have these // marshaled on every status call. if w.updateTags.Load() { - tags = w.tags.Load().(map[string]*servers.TagValues) + tags = w.tags.Load().([]*servers.TagPair) } statusCtx, statusCancel := context.WithTimeout(cancelCtx, common.StatusTimeout) defer statusCancel() result, err := client.Status(statusCtx, &pbs.StatusRequest{ Jobs: activeJobs, - Worker: &servers.Server{ - PrivateId: w.conf.RawConfig.Worker.Name, - Address: w.conf.RawConfig.Worker.PublicAddr, - Tags: tags, + WorkerStatus: &servers.ServerWorkerStatus{ + Name: w.conf.RawConfig.Worker.Name, + Address: w.conf.RawConfig.Worker.PublicAddr, + Tags: tags, }, UpdateTags: w.updateTags.Load(), }) @@ -190,9 +190,9 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) { } else { w.updateTags.Store(false) // This may be nil if we are in a multiple hop scenario - if len(result.Controllers) > 0 { - addrs := make([]resolver.Address, 0, len(result.Controllers)) - for _, v := range result.Controllers { + if len(result.CalculatedUpstreams) > 0 { + addrs := make([]resolver.Address, 0, len(result.CalculatedUpstreams)) + for _, v := range result.CalculatedUpstreams { addrs = append(addrs, resolver.Address{Addr: v.Address}) } w.Resolver().UpdateState(resolver.State{Addresses: addrs}) diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index d73145cc9..623cbd0d7 100644 --- a/internal/daemon/worker/testing.go +++ b/internal/daemon/worker/testing.go @@ -56,7 +56,7 @@ func (tw *TestWorker) Name() string { func (tw *TestWorker) ControllerAddrs() []string { var addrs []string lastStatus := tw.w.LastStatusSuccess() - for _, v := range lastStatus.GetControllers() { + for _, v := range lastStatus.GetCalculatedUpstreams() { addrs = append(addrs, v.Address) } diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index 5cca1f524..26dc8c0dd 100644 --- a/internal/daemon/worker/worker.go +++ b/internal/daemon/worker/worker.go @@ -294,13 +294,16 @@ func (w *Worker) Resolver() *manual.Resolver { func (w *Worker) ParseAndStoreTags(incoming map[string][]string) { if len(incoming) == 0 { - w.tags.Store(map[string]*servers.TagValues{}) + w.tags.Store([]*servers.TagPair{}) return } - tags := make(map[string]*servers.TagValues, len(incoming)) - for k, v := range incoming { - tags[k] = &servers.TagValues{ - Values: append(make([]string, 0, len(v)), v...), + tags := []*servers.TagPair{} + for k, vals := range incoming { + for _, v := range vals { + tags = append(tags, &servers.TagPair{ + Key: k, + Value: v, + }) } } w.tags.Store(tags) diff --git a/internal/daemon/worker/worker_proxy_service.go b/internal/daemon/worker/worker_proxy_service.go index e70b4451a..1b4755658 100644 --- a/internal/daemon/worker/worker_proxy_service.go +++ b/internal/daemon/worker/worker_proxy_service.go @@ -36,7 +36,7 @@ func (ws *workerProxyServiceServer) Status(ctx context.Context, req *pbs.StatusR if resp != nil { // We don't currently support distributing new addreses to workers // multiple hops away so ensure they're stripped out - resp.Controllers = nil + resp.CalculatedUpstreams = nil } return resp, err diff --git a/internal/gen/controller/servers/services/server_coordination_service.pb.go b/internal/gen/controller/servers/services/server_coordination_service.pb.go index c893c7c2b..8fc3654e2 100644 --- a/internal/gen/controller/servers/services/server_coordination_service.pb.go +++ b/internal/gen/controller/servers/services/server_coordination_service.pb.go @@ -222,6 +222,55 @@ func (CHANGETYPE) EnumDescriptor() ([]byte, []int) { return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{3} } +type UpstreamServer_TYPE int32 + +const ( + UpstreamServer_TYPE_UNSPECIFIED UpstreamServer_TYPE = 0 + UpstreamServer_TYPE_CONTROLLER UpstreamServer_TYPE = 1 + UpstreamServer_TYPE_WORKER UpstreamServer_TYPE = 2 +) + +// Enum value maps for UpstreamServer_TYPE. +var ( + UpstreamServer_TYPE_name = map[int32]string{ + 0: "TYPE_UNSPECIFIED", + 1: "TYPE_CONTROLLER", + 2: "TYPE_WORKER", + } + UpstreamServer_TYPE_value = map[string]int32{ + "TYPE_UNSPECIFIED": 0, + "TYPE_CONTROLLER": 1, + "TYPE_WORKER": 2, + } +) + +func (x UpstreamServer_TYPE) Enum() *UpstreamServer_TYPE { + p := new(UpstreamServer_TYPE) + *p = x + return p +} + +func (x UpstreamServer_TYPE) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (UpstreamServer_TYPE) Descriptor() protoreflect.EnumDescriptor { + return file_controller_servers_services_v1_server_coordination_service_proto_enumTypes[4].Descriptor() +} + +func (UpstreamServer_TYPE) Type() protoreflect.EnumType { + return &file_controller_servers_services_v1_server_coordination_service_proto_enumTypes[4] +} + +func (x UpstreamServer_TYPE) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use UpstreamServer_TYPE.Descriptor instead. +func (UpstreamServer_TYPE) EnumDescriptor() ([]byte, []int) { + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{4, 0} +} + type Connection struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -462,15 +511,70 @@ func (x *JobStatus) GetJob() *Job { return nil } +// UpstreamServer returns the controller address in the StatusResponse but may be used in the future to return +// worker address. +type UpstreamServer struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // type specifies between controller or worker. + Type UpstreamServer_TYPE `protobuf:"varint,10,opt,name=type,proto3,enum=controller.servers.services.v1.UpstreamServer_TYPE" json:"type,omitempty"` + // Address of the daemon + Address string `protobuf:"bytes,20,opt,name=address,proto3" json:"address,omitempty"` +} + +func (x *UpstreamServer) Reset() { + *x = UpstreamServer{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *UpstreamServer) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UpstreamServer) ProtoMessage() {} + +func (x *UpstreamServer) ProtoReflect() protoreflect.Message { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UpstreamServer.ProtoReflect.Descriptor instead. +func (*UpstreamServer) Descriptor() ([]byte, []int) { + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{4} +} + +func (x *UpstreamServer) GetType() UpstreamServer_TYPE { + if x != nil { + return x.Type + } + return UpstreamServer_TYPE_UNSPECIFIED +} + +func (x *UpstreamServer) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + type StatusRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // The worker info. We could use information from the TLS connection but this - // is easier and going the other route doesn't provide much benefit -- if you - // get access to the key and spoof the connection, you're already compromised. - Worker *servers.Server `protobuf:"bytes,10,opt,name=worker,proto3" json:"worker,omitempty"` // Jobs which this worker wants to report the status. Jobs []*JobStatus `protobuf:"bytes,20,rep,name=jobs,proto3" json:"jobs,omitempty"` // Whether to update tags from the Server block on this RPC. We only need to @@ -478,12 +582,17 @@ type StatusRequest struct { // changed allows us to avoid constant database operations for something that // won't change very often, if ever. UpdateTags bool `protobuf:"varint,30,opt,name=update_tags,json=updateTags,proto3" json:"update_tags,omitempty"` + // Replaces the old worker field: + // The worker info. We could use information from the TLS connection but this + // is easier and going the other route doesn't provide much benefit -- if you + // get access to the key and spoof the connection, you're already compromised. + WorkerStatus *servers.ServerWorkerStatus `protobuf:"bytes,40,opt,name=worker_status,json=workerStatus,proto3" json:"worker_status,omitempty"` } func (x *StatusRequest) Reset() { *x = StatusRequest{} if protoimpl.UnsafeEnabled { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[4] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -496,7 +605,7 @@ func (x *StatusRequest) String() string { func (*StatusRequest) ProtoMessage() {} func (x *StatusRequest) ProtoReflect() protoreflect.Message { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[4] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[5] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -509,14 +618,7 @@ func (x *StatusRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use StatusRequest.ProtoReflect.Descriptor instead. func (*StatusRequest) Descriptor() ([]byte, []int) { - return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{4} -} - -func (x *StatusRequest) GetWorker() *servers.Server { - if x != nil { - return x.Worker - } - return nil + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{5} } func (x *StatusRequest) GetJobs() []*JobStatus { @@ -533,6 +635,13 @@ func (x *StatusRequest) GetUpdateTags() bool { return false } +func (x *StatusRequest) GetWorkerStatus() *servers.ServerWorkerStatus { + if x != nil { + return x.WorkerStatus + } + return nil +} + type JobChangeRequest struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -545,7 +654,7 @@ type JobChangeRequest struct { func (x *JobChangeRequest) Reset() { *x = JobChangeRequest{} if protoimpl.UnsafeEnabled { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[5] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -558,7 +667,7 @@ func (x *JobChangeRequest) String() string { func (*JobChangeRequest) ProtoMessage() {} func (x *JobChangeRequest) ProtoReflect() protoreflect.Message { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[5] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[6] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -571,7 +680,7 @@ func (x *JobChangeRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use JobChangeRequest.ProtoReflect.Descriptor instead. func (*JobChangeRequest) Descriptor() ([]byte, []int) { - return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{5} + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{6} } func (x *JobChangeRequest) GetJob() *Job { @@ -593,20 +702,23 @@ type StatusResponse struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // Active controllers. This can be used for connection management. - Controllers []*servers.Server `protobuf:"bytes,10,rep,name=controllers,proto3" json:"controllers,omitempty"` // List of jobs and the expected state changes. For example, this will // include jobs witch change type of canceled for jobs which are active on a // worker but should be canceled. This could also contain a request to start a // job such as a worker -> worker proxy for establishing a session through an // enclave. JobsRequests []*JobChangeRequest `protobuf:"bytes,20,rep,name=jobs_requests,json=jobsRequests,proto3" json:"jobs_requests,omitempty"` + // UpstreamServer currently returns the controller address in the StatusResponse. + CalculatedUpstreams []*UpstreamServer `protobuf:"bytes,30,rep,name=calculated_upstreams,json=calculatedUpstreams,proto3" json:"calculated_upstreams,omitempty"` + // The ID of the worker which made the request. The worker can send this value in subsequent requests so the + // controller does not need to do a database lookup for the id using the name field. + WorkerId string `protobuf:"bytes,40,opt,name=worker_id,json=workerId,proto3" json:"worker_id,omitempty"` } func (x *StatusResponse) Reset() { *x = StatusResponse{} if protoimpl.UnsafeEnabled { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[6] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -619,7 +731,7 @@ func (x *StatusResponse) String() string { func (*StatusResponse) ProtoMessage() {} func (x *StatusResponse) ProtoReflect() protoreflect.Message { - mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[6] + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[7] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -632,14 +744,7 @@ func (x *StatusResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use StatusResponse.ProtoReflect.Descriptor instead. func (*StatusResponse) Descriptor() ([]byte, []int) { - return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{6} -} - -func (x *StatusResponse) GetControllers() []*servers.Server { - if x != nil { - return x.Controllers - } - return nil + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{7} } func (x *StatusResponse) GetJobsRequests() []*JobChangeRequest { @@ -649,6 +754,20 @@ func (x *StatusResponse) GetJobsRequests() []*JobChangeRequest { return nil } +func (x *StatusResponse) GetCalculatedUpstreams() []*UpstreamServer { + if x != nil { + return x.CalculatedUpstreams + } + return nil +} + +func (x *StatusResponse) GetWorkerId() string { + if x != nil { + return x.WorkerId + } + return "" +} + var File_controller_servers_services_v1_server_coordination_service_proto protoreflect.FileDescriptor var file_controller_servers_services_v1_server_coordination_service_proto_rawDesc = []byte{ @@ -695,79 +814,99 @@ var file_controller_servers_services_v1_server_coordination_service_proto_rawDes 0x0a, 0x03, 0x6a, 0x6f, 0x62, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, - 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x22, 0xa6, 0x01, 0x0a, 0x0d, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, - 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x35, 0x0a, 0x06, 0x77, 0x6f, 0x72, 0x6b, 0x65, - 0x72, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, - 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x2e, - 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x52, 0x06, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x3d, - 0x0a, 0x04, 0x6a, 0x6f, 0x62, 0x73, 0x18, 0x14, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x29, 0x2e, 0x63, - 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, - 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, - 0x62, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x04, 0x6a, 0x6f, 0x62, 0x73, 0x12, 0x1f, 0x0a, - 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x61, 0x67, 0x73, 0x18, 0x1e, 0x20, 0x01, - 0x28, 0x08, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x61, 0x67, 0x73, 0x22, 0x98, - 0x01, 0x0a, 0x10, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, - 0x65, 0x73, 0x74, 0x12, 0x35, 0x0a, 0x03, 0x6a, 0x6f, 0x62, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, - 0x32, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, - 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, - 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x12, 0x4d, 0x0a, 0x0c, 0x72, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, - 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, - 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, - 0x31, 0x2e, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x52, 0x0b, 0x72, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x22, 0xa8, 0x01, 0x0a, 0x0e, 0x53, 0x74, - 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x3f, 0x0a, 0x0b, - 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x18, 0x0a, 0x20, 0x03, 0x28, - 0x0b, 0x32, 0x1d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, - 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, - 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x12, 0x55, 0x0a, + 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x22, 0xb7, 0x01, 0x0a, 0x0e, 0x55, 0x70, 0x73, 0x74, 0x72, 0x65, + 0x61, 0x6d, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x12, 0x47, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, + 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x33, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, + 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, + 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x55, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x54, 0x59, 0x50, 0x45, 0x52, 0x04, 0x74, 0x79, 0x70, + 0x65, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x18, 0x14, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x22, 0x42, 0x0a, 0x04, 0x54, + 0x59, 0x50, 0x45, 0x12, 0x14, 0x0a, 0x10, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, + 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x13, 0x0a, 0x0f, 0x54, 0x59, 0x50, + 0x45, 0x5f, 0x43, 0x4f, 0x4e, 0x54, 0x52, 0x4f, 0x4c, 0x4c, 0x45, 0x52, 0x10, 0x01, 0x12, 0x0f, + 0x0a, 0x0b, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x57, 0x4f, 0x52, 0x4b, 0x45, 0x52, 0x10, 0x02, 0x22, + 0xcd, 0x01, 0x0a, 0x0d, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x12, 0x3d, 0x0a, 0x04, 0x6a, 0x6f, 0x62, 0x73, 0x18, 0x14, 0x20, 0x03, 0x28, 0x0b, 0x32, + 0x29, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, + 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, + 0x2e, 0x4a, 0x6f, 0x62, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x04, 0x6a, 0x6f, 0x62, 0x73, + 0x12, 0x1f, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x61, 0x67, 0x73, 0x18, + 0x1e, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x61, 0x67, + 0x73, 0x12, 0x4e, 0x0a, 0x0d, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x73, 0x74, 0x61, 0x74, + 0x75, 0x73, 0x18, 0x28, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x29, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, + 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, + 0x2e, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, + 0x74, 0x75, 0x73, 0x52, 0x0c, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x75, + 0x73, 0x4a, 0x04, 0x08, 0x0a, 0x10, 0x0b, 0x52, 0x06, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x22, + 0x98, 0x01, 0x0a, 0x10, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x12, 0x35, 0x0a, 0x03, 0x6a, 0x6f, 0x62, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x0b, 0x32, 0x23, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, + 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, + 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x52, 0x03, 0x6a, 0x6f, 0x62, 0x12, 0x4d, 0x0a, 0x0c, 0x72, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, + 0x0e, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, + 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, + 0x76, 0x31, 0x2e, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x52, 0x0b, 0x72, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x22, 0xfa, 0x01, 0x0a, 0x0e, 0x53, + 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x55, 0x0a, 0x0d, 0x6a, 0x6f, 0x62, 0x73, 0x5f, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x73, 0x18, 0x14, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x4a, 0x6f, 0x62, 0x43, 0x68, 0x61, 0x6e, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x0c, 0x6a, 0x6f, 0x62, 0x73, 0x52, 0x65, 0x71, 0x75, - 0x65, 0x73, 0x74, 0x73, 0x2a, 0x92, 0x01, 0x0a, 0x10, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, - 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x20, 0x0a, 0x1c, 0x43, 0x4f, 0x4e, - 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, - 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1f, 0x0a, 0x1b, 0x43, - 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, - 0x41, 0x55, 0x54, 0x48, 0x4f, 0x52, 0x49, 0x5a, 0x45, 0x44, 0x10, 0x01, 0x12, 0x1e, 0x0a, 0x1a, - 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, - 0x5f, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x45, 0x44, 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17, - 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, - 0x5f, 0x43, 0x4c, 0x4f, 0x53, 0x45, 0x44, 0x10, 0x03, 0x2a, 0x9e, 0x01, 0x0a, 0x0d, 0x53, 0x45, - 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x1d, 0x0a, 0x19, 0x53, - 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x55, 0x4e, 0x53, - 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x19, 0x0a, 0x15, 0x53, 0x45, - 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, 0x45, 0x4e, 0x44, - 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x18, 0x0a, 0x14, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, - 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x41, 0x43, 0x54, 0x49, 0x56, 0x45, 0x10, 0x02, 0x12, - 0x1b, 0x0a, 0x17, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, - 0x5f, 0x43, 0x41, 0x4e, 0x43, 0x45, 0x4c, 0x49, 0x4e, 0x47, 0x10, 0x03, 0x12, 0x1c, 0x0a, 0x18, - 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x54, 0x45, - 0x52, 0x4d, 0x49, 0x4e, 0x41, 0x54, 0x45, 0x44, 0x10, 0x04, 0x2a, 0x37, 0x0a, 0x07, 0x4a, 0x4f, - 0x42, 0x54, 0x59, 0x50, 0x45, 0x12, 0x17, 0x0a, 0x13, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, - 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x13, - 0x0a, 0x0f, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, - 0x4e, 0x10, 0x01, 0x2a, 0x45, 0x0a, 0x0a, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, - 0x45, 0x12, 0x1a, 0x0a, 0x16, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, - 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1b, 0x0a, - 0x17, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x50, 0x44, 0x41, - 0x54, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x54, 0x45, 0x10, 0x01, 0x32, 0x86, 0x01, 0x0a, 0x19, 0x53, - 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6f, 0x72, 0x64, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, - 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x69, 0x0a, 0x06, 0x53, 0x74, 0x61, 0x74, - 0x75, 0x73, 0x12, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, + 0x65, 0x73, 0x74, 0x73, 0x12, 0x61, 0x0a, 0x14, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c, 0x61, 0x74, + 0x65, 0x64, 0x5f, 0x75, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x18, 0x1e, 0x20, 0x03, + 0x28, 0x0b, 0x32, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, - 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, - 0x74, 0x1a, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, - 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2e, - 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, - 0x65, 0x22, 0x00, 0x42, 0x51, 0x5a, 0x4f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, - 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, 0x75, 0x6e, - 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x67, 0x65, - 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x65, 0x72, - 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x3b, 0x73, 0x65, - 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x2e, 0x76, 0x31, 0x2e, 0x55, 0x70, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x53, 0x65, 0x72, 0x76, + 0x65, 0x72, 0x52, 0x13, 0x63, 0x61, 0x6c, 0x63, 0x75, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x55, 0x70, + 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x77, 0x6f, 0x72, 0x6b, 0x65, + 0x72, 0x5f, 0x69, 0x64, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x77, 0x6f, 0x72, 0x6b, + 0x65, 0x72, 0x49, 0x64, 0x4a, 0x04, 0x08, 0x0a, 0x10, 0x0b, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, + 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x2a, 0x92, 0x01, 0x0a, 0x10, 0x43, 0x4f, 0x4e, 0x4e, + 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x20, 0x0a, 0x1c, + 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, + 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1f, + 0x0a, 0x1b, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, + 0x55, 0x53, 0x5f, 0x41, 0x55, 0x54, 0x48, 0x4f, 0x52, 0x49, 0x5a, 0x45, 0x44, 0x10, 0x01, 0x12, + 0x1e, 0x0a, 0x1a, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, + 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x45, 0x44, 0x10, 0x02, 0x12, + 0x1b, 0x0a, 0x17, 0x43, 0x4f, 0x4e, 0x4e, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, + 0x54, 0x55, 0x53, 0x5f, 0x43, 0x4c, 0x4f, 0x53, 0x45, 0x44, 0x10, 0x03, 0x2a, 0x9e, 0x01, 0x0a, + 0x0d, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x12, 0x1d, + 0x0a, 0x19, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, + 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x19, 0x0a, + 0x15, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x50, + 0x45, 0x4e, 0x44, 0x49, 0x4e, 0x47, 0x10, 0x01, 0x12, 0x18, 0x0a, 0x14, 0x53, 0x45, 0x53, 0x53, + 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, 0x5f, 0x41, 0x43, 0x54, 0x49, 0x56, 0x45, + 0x10, 0x02, 0x12, 0x1b, 0x0a, 0x17, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, + 0x54, 0x55, 0x53, 0x5f, 0x43, 0x41, 0x4e, 0x43, 0x45, 0x4c, 0x49, 0x4e, 0x47, 0x10, 0x03, 0x12, + 0x1c, 0x0a, 0x18, 0x53, 0x45, 0x53, 0x53, 0x49, 0x4f, 0x4e, 0x53, 0x54, 0x41, 0x54, 0x55, 0x53, + 0x5f, 0x54, 0x45, 0x52, 0x4d, 0x49, 0x4e, 0x41, 0x54, 0x45, 0x44, 0x10, 0x04, 0x2a, 0x37, 0x0a, + 0x07, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x12, 0x17, 0x0a, 0x13, 0x4a, 0x4f, 0x42, 0x54, + 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, + 0x00, 0x12, 0x13, 0x0a, 0x0f, 0x4a, 0x4f, 0x42, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x45, 0x53, + 0x53, 0x49, 0x4f, 0x4e, 0x10, 0x01, 0x2a, 0x45, 0x0a, 0x0a, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, + 0x54, 0x59, 0x50, 0x45, 0x12, 0x1a, 0x0a, 0x16, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, + 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, + 0x12, 0x1b, 0x0a, 0x17, 0x43, 0x48, 0x41, 0x4e, 0x47, 0x45, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, + 0x50, 0x44, 0x41, 0x54, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x54, 0x45, 0x10, 0x01, 0x32, 0x86, 0x01, + 0x0a, 0x19, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6f, 0x72, 0x64, 0x69, 0x6e, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x69, 0x0a, 0x06, 0x53, + 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x2d, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, + 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, + 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, + 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, + 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, + 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x51, 0x5a, 0x4f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, + 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, + 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, + 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, + 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, + 0x3b, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x33, } var ( @@ -782,42 +921,45 @@ func file_controller_servers_services_v1_server_coordination_service_proto_rawDe return file_controller_servers_services_v1_server_coordination_service_proto_rawDescData } -var file_controller_servers_services_v1_server_coordination_service_proto_enumTypes = make([]protoimpl.EnumInfo, 4) -var file_controller_servers_services_v1_server_coordination_service_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_controller_servers_services_v1_server_coordination_service_proto_enumTypes = make([]protoimpl.EnumInfo, 5) +var file_controller_servers_services_v1_server_coordination_service_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_controller_servers_services_v1_server_coordination_service_proto_goTypes = []interface{}{ - (CONNECTIONSTATUS)(0), // 0: controller.servers.services.v1.CONNECTIONSTATUS - (SESSIONSTATUS)(0), // 1: controller.servers.services.v1.SESSIONSTATUS - (JOBTYPE)(0), // 2: controller.servers.services.v1.JOBTYPE - (CHANGETYPE)(0), // 3: controller.servers.services.v1.CHANGETYPE - (*Connection)(nil), // 4: controller.servers.services.v1.Connection - (*SessionJobInfo)(nil), // 5: controller.servers.services.v1.SessionJobInfo - (*Job)(nil), // 6: controller.servers.services.v1.Job - (*JobStatus)(nil), // 7: controller.servers.services.v1.JobStatus - (*StatusRequest)(nil), // 8: controller.servers.services.v1.StatusRequest - (*JobChangeRequest)(nil), // 9: controller.servers.services.v1.JobChangeRequest - (*StatusResponse)(nil), // 10: controller.servers.services.v1.StatusResponse - (*servers.Server)(nil), // 11: controller.servers.v1.Server + (CONNECTIONSTATUS)(0), // 0: controller.servers.services.v1.CONNECTIONSTATUS + (SESSIONSTATUS)(0), // 1: controller.servers.services.v1.SESSIONSTATUS + (JOBTYPE)(0), // 2: controller.servers.services.v1.JOBTYPE + (CHANGETYPE)(0), // 3: controller.servers.services.v1.CHANGETYPE + (UpstreamServer_TYPE)(0), // 4: controller.servers.services.v1.UpstreamServer.TYPE + (*Connection)(nil), // 5: controller.servers.services.v1.Connection + (*SessionJobInfo)(nil), // 6: controller.servers.services.v1.SessionJobInfo + (*Job)(nil), // 7: controller.servers.services.v1.Job + (*JobStatus)(nil), // 8: controller.servers.services.v1.JobStatus + (*UpstreamServer)(nil), // 9: controller.servers.services.v1.UpstreamServer + (*StatusRequest)(nil), // 10: controller.servers.services.v1.StatusRequest + (*JobChangeRequest)(nil), // 11: controller.servers.services.v1.JobChangeRequest + (*StatusResponse)(nil), // 12: controller.servers.services.v1.StatusResponse + (*servers.ServerWorkerStatus)(nil), // 13: controller.servers.v1.ServerWorkerStatus } var file_controller_servers_services_v1_server_coordination_service_proto_depIdxs = []int32{ 0, // 0: controller.servers.services.v1.Connection.status:type_name -> controller.servers.services.v1.CONNECTIONSTATUS 1, // 1: controller.servers.services.v1.SessionJobInfo.status:type_name -> controller.servers.services.v1.SESSIONSTATUS - 4, // 2: controller.servers.services.v1.SessionJobInfo.connections:type_name -> controller.servers.services.v1.Connection + 5, // 2: controller.servers.services.v1.SessionJobInfo.connections:type_name -> controller.servers.services.v1.Connection 2, // 3: controller.servers.services.v1.Job.type:type_name -> controller.servers.services.v1.JOBTYPE - 5, // 4: controller.servers.services.v1.Job.session_info:type_name -> controller.servers.services.v1.SessionJobInfo - 6, // 5: controller.servers.services.v1.JobStatus.job:type_name -> controller.servers.services.v1.Job - 11, // 6: controller.servers.services.v1.StatusRequest.worker:type_name -> controller.servers.v1.Server - 7, // 7: controller.servers.services.v1.StatusRequest.jobs:type_name -> controller.servers.services.v1.JobStatus - 6, // 8: controller.servers.services.v1.JobChangeRequest.job:type_name -> controller.servers.services.v1.Job - 3, // 9: controller.servers.services.v1.JobChangeRequest.request_type:type_name -> controller.servers.services.v1.CHANGETYPE - 11, // 10: controller.servers.services.v1.StatusResponse.controllers:type_name -> controller.servers.v1.Server - 9, // 11: controller.servers.services.v1.StatusResponse.jobs_requests:type_name -> controller.servers.services.v1.JobChangeRequest - 8, // 12: controller.servers.services.v1.ServerCoordinationService.Status:input_type -> controller.servers.services.v1.StatusRequest - 10, // 13: controller.servers.services.v1.ServerCoordinationService.Status:output_type -> controller.servers.services.v1.StatusResponse - 13, // [13:14] is the sub-list for method output_type - 12, // [12:13] is the sub-list for method input_type - 12, // [12:12] is the sub-list for extension type_name - 12, // [12:12] is the sub-list for extension extendee - 0, // [0:12] is the sub-list for field type_name + 6, // 4: controller.servers.services.v1.Job.session_info:type_name -> controller.servers.services.v1.SessionJobInfo + 7, // 5: controller.servers.services.v1.JobStatus.job:type_name -> controller.servers.services.v1.Job + 4, // 6: controller.servers.services.v1.UpstreamServer.type:type_name -> controller.servers.services.v1.UpstreamServer.TYPE + 8, // 7: controller.servers.services.v1.StatusRequest.jobs:type_name -> controller.servers.services.v1.JobStatus + 13, // 8: controller.servers.services.v1.StatusRequest.worker_status:type_name -> controller.servers.v1.ServerWorkerStatus + 7, // 9: controller.servers.services.v1.JobChangeRequest.job:type_name -> controller.servers.services.v1.Job + 3, // 10: controller.servers.services.v1.JobChangeRequest.request_type:type_name -> controller.servers.services.v1.CHANGETYPE + 11, // 11: controller.servers.services.v1.StatusResponse.jobs_requests:type_name -> controller.servers.services.v1.JobChangeRequest + 9, // 12: controller.servers.services.v1.StatusResponse.calculated_upstreams:type_name -> controller.servers.services.v1.UpstreamServer + 10, // 13: controller.servers.services.v1.ServerCoordinationService.Status:input_type -> controller.servers.services.v1.StatusRequest + 12, // 14: controller.servers.services.v1.ServerCoordinationService.Status:output_type -> controller.servers.services.v1.StatusResponse + 14, // [14:15] is the sub-list for method output_type + 13, // [13:14] is the sub-list for method input_type + 13, // [13:13] is the sub-list for extension type_name + 13, // [13:13] is the sub-list for extension extendee + 0, // [0:13] is the sub-list for field type_name } func init() { file_controller_servers_services_v1_server_coordination_service_proto_init() } @@ -875,7 +1017,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( } } file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*StatusRequest); i { + switch v := v.(*UpstreamServer); i { case 0: return &v.state case 1: @@ -887,7 +1029,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( } } file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*JobChangeRequest); i { + switch v := v.(*StatusRequest); i { case 0: return &v.state case 1: @@ -899,6 +1041,18 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( } } file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*JobChangeRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*StatusResponse); i { case 0: return &v.state @@ -919,8 +1073,8 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_controller_servers_services_v1_server_coordination_service_proto_rawDesc, - NumEnums: 4, - NumMessages: 7, + NumEnums: 5, + NumMessages: 8, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/proto/controller/servers/services/v1/server_coordination_service.proto b/internal/proto/controller/servers/services/v1/server_coordination_service.proto index f571efe4a..6f47f3918 100644 --- a/internal/proto/controller/servers/services/v1/server_coordination_service.proto +++ b/internal/proto/controller/servers/services/v1/server_coordination_service.proto @@ -56,11 +56,25 @@ message JobStatus { Job job = 1; } +// UpstreamServer returns the controller address in the StatusResponse but may be used in the future to return +// worker address. +message UpstreamServer { + enum TYPE { + TYPE_UNSPECIFIED = 0; + TYPE_CONTROLLER = 1; + TYPE_WORKER = 2; + } + + // type specifies between controller or worker. + TYPE type = 10; + + // Address of the daemon + string address = 20; +} + message StatusRequest { - // The worker info. We could use information from the TLS connection but this - // is easier and going the other route doesn't provide much benefit -- if you - // get access to the key and spoof the connection, you're already compromised. - servers.v1.Server worker = 10; + reserved 10; + reserved "worker"; // Jobs which this worker wants to report the status. repeated JobStatus jobs = 20; @@ -70,6 +84,12 @@ message StatusRequest { // changed allows us to avoid constant database operations for something that // won't change very often, if ever. bool update_tags = 30; + + // Replaces the old worker field: + // The worker info. We could use information from the TLS connection but this + // is easier and going the other route doesn't provide much benefit -- if you + // get access to the key and spoof the connection, you're already compromised. + servers.v1.ServerWorkerStatus worker_status = 40; } enum CHANGETYPE { @@ -85,8 +105,8 @@ message JobChangeRequest { } message StatusResponse { - // Active controllers. This can be used for connection management. - repeated servers.v1.Server controllers = 10; + reserved 10; + reserved "controllers"; // List of jobs and the expected state changes. For example, this will // include jobs witch change type of canceled for jobs which are active on a @@ -94,4 +114,11 @@ message StatusResponse { // job such as a worker -> worker proxy for establishing a session through an // enclave. repeated JobChangeRequest jobs_requests = 20; + + // UpstreamServer currently returns the controller address in the StatusResponse. + repeated UpstreamServer calculated_upstreams = 30; + + // The ID of the worker which made the request. The worker can send this value in subsequent requests so the + // controller does not need to do a database lookup for the id using the name field. + string worker_id = 40; } diff --git a/internal/proto/controller/servers/v1/servers.proto b/internal/proto/controller/servers/v1/servers.proto index 493db240a..3339c9ba9 100644 --- a/internal/proto/controller/servers/v1/servers.proto +++ b/internal/proto/controller/servers/v1/servers.proto @@ -2,44 +2,25 @@ syntax = "proto3"; package controller.servers.v1; -import "controller/storage/timestamp/v1/timestamp.proto"; - option go_package = "github.com/hashicorp/boundary/internal/servers;servers"; -// Server contains all fields related to a Controller or Worker resource -message Server { - // Private ID of the resource - string private_id = 10; +// TagPair matches a key to a value. +message TagPair { + string key = 1; + string value = 2; +} - // Type of the resource (controller, worker) - string type = 20; +// ServerWorkerStatus is the new message used in place of Server to relay status request info. +message ServerWorkerStatus { + // Id of the worker. + string public_id = 10; - // Name of the resource. This is deprecated because we ended up setting the - // private ID and name to be identical, and there is currently no reason we - // can think of to allow a given server's name to match another within the - // same cluster. So we simply use the private ID and deprecate this. - // @inject_tag: `gorm:"-"` - string name = 30 [deprecated = true]; + // Name of the resource (optional) + string name = 20; - // Description of the resource - string description = 40; - - // Address for the server - string address = 50; - - // First seen time from the RDBMS - storage.timestamp.v1.Timestamp create_time = 60; - - // Last time there was an update - storage.timestamp.v1.Timestamp update_time = 70; + // Address for the worker. This must be set. + string address = 30; // Tags for workers - // @inject_tag: `gorm:"-"` - map tags = 80; -} - -// TagValues is used because map fields cannot be repeated but can be a -// message -message TagValues { - repeated string values = 10; + repeated TagPair tags = 40; } diff --git a/internal/servers/options.go b/internal/servers/options.go index 999782588..052ed55f5 100644 --- a/internal/servers/options.go +++ b/internal/servers/options.go @@ -22,6 +22,7 @@ type Option func(*options) // options = how options are represented type options struct { withName string + withPublicId string withDescription string withAddress string withLimit int @@ -57,6 +58,13 @@ func WithName(name string) Option { } } +// WithPublicId provides an optional public Id used for skipping one db call. +func WithPublicId(id string) Option { + return func(o *options) { + o.withPublicId = id + } +} + // WithAddress provides an optional address. func WithAddress(address string) Option { return func(o *options) { diff --git a/internal/servers/options_test.go b/internal/servers/options_test.go index 3a34e4a7d..a1ef9ad8e 100644 --- a/internal/servers/options_test.go +++ b/internal/servers/options_test.go @@ -21,6 +21,14 @@ func Test_GetOpts(t *testing.T) { testOpts.withNewIdFunc = nil assert.Equal(t, opts, testOpts) }) + t.Run("WithPublicId", func(t *testing.T) { + opts := getOpts(WithPublicId("test")) + testOpts := getDefaultOptions() + testOpts.withPublicId = "test" + opts.withNewIdFunc = nil + testOpts.withNewIdFunc = nil + assert.Equal(t, opts, testOpts) + }) t.Run("WithDescription", func(t *testing.T) { opts := getOpts(WithDescription("test desc")) testOpts := getDefaultOptions() diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index e6f20568d..c06fba951 100644 --- a/internal/servers/repository_worker.go +++ b/internal/servers/repository_worker.go @@ -178,7 +178,7 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ... // UpsertWorkerStatus creates a new worker if one with the provided public id // doesn't already exist. If it does, UpsertWorkerStatus updates the worker // status. This returns the Worker object with the updated WorkerStatus applied. -// The WithUpdateTags option is the only ones used. All others are ignored. +// The WithPublicId and WithUpdateTags options are the only ones used. All others are ignored. // Workers are intentionally not oplogged. func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt ...Option) (*Worker, error) { const op = "servers.UpsertWorkerStatus" @@ -207,9 +207,14 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt opts := getOpts(opt...) var err error - worker.PublicId, err = newWorkerIdFromName(ctx, worker.GetWorkerReportedName()) - if err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error generating worker id")) + switch { + case opts.withPublicId != "": + worker.PublicId = opts.withPublicId + default: + worker.PublicId, err = newWorkerIdFromName(ctx, worker.GetWorkerReportedName()) + if err != nil { + return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error generating worker id")) + } } var ret *Worker diff --git a/internal/servers/servers.pb.go b/internal/servers/servers.pb.go index 22d492e8a..46f2879cc 100644 --- a/internal/servers/servers.pb.go +++ b/internal/servers/servers.pb.go @@ -7,7 +7,6 @@ package servers import ( - timestamp "github.com/hashicorp/boundary/internal/db/timestamp" protoreflect "google.golang.org/protobuf/reflect/protoreflect" protoimpl "google.golang.org/protobuf/runtime/protoimpl" reflect "reflect" @@ -21,39 +20,18 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// Server contains all fields related to a Controller or Worker resource -type Server struct { +// TagPair matches a key to a value. +type TagPair struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - // Private ID of the resource - PrivateId string `protobuf:"bytes,10,opt,name=private_id,json=privateId,proto3" json:"private_id,omitempty"` - // Type of the resource (controller, worker) - Type string `protobuf:"bytes,20,opt,name=type,proto3" json:"type,omitempty"` - // Name of the resource. This is deprecated because we ended up setting the - // private ID and name to be identical, and there is currently no reason we - // can think of to allow a given server's name to match another within the - // same cluster. So we simply use the private ID and deprecate this. - // @inject_tag: `gorm:"-"` - // - // Deprecated: Do not use. - Name string `protobuf:"bytes,30,opt,name=name,proto3" json:"name,omitempty" gorm:"-"` - // Description of the resource - Description string `protobuf:"bytes,40,opt,name=description,proto3" json:"description,omitempty"` - // Address for the server - Address string `protobuf:"bytes,50,opt,name=address,proto3" json:"address,omitempty"` - // First seen time from the RDBMS - CreateTime *timestamp.Timestamp `protobuf:"bytes,60,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty"` - // Last time there was an update - UpdateTime *timestamp.Timestamp `protobuf:"bytes,70,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty"` - // Tags for workers - // @inject_tag: `gorm:"-"` - Tags map[string]*TagValues `protobuf:"bytes,80,rep,name=tags,proto3" json:"tags,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3" gorm:"-"` + Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` + Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"` } -func (x *Server) Reset() { - *x = Server{} +func (x *TagPair) Reset() { + *x = TagPair{} if protoimpl.UnsafeEnabled { mi := &file_controller_servers_v1_servers_proto_msgTypes[0] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -61,13 +39,13 @@ func (x *Server) Reset() { } } -func (x *Server) String() string { +func (x *TagPair) String() string { return protoimpl.X.MessageStringOf(x) } -func (*Server) ProtoMessage() {} +func (*TagPair) ProtoMessage() {} -func (x *Server) ProtoReflect() protoreflect.Message { +func (x *TagPair) ProtoReflect() protoreflect.Message { mi := &file_controller_servers_v1_servers_proto_msgTypes[0] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -79,80 +57,43 @@ func (x *Server) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use Server.ProtoReflect.Descriptor instead. -func (*Server) Descriptor() ([]byte, []int) { +// Deprecated: Use TagPair.ProtoReflect.Descriptor instead. +func (*TagPair) Descriptor() ([]byte, []int) { return file_controller_servers_v1_servers_proto_rawDescGZIP(), []int{0} } -func (x *Server) GetPrivateId() string { +func (x *TagPair) GetKey() string { if x != nil { - return x.PrivateId + return x.Key } return "" } -func (x *Server) GetType() string { +func (x *TagPair) GetValue() string { if x != nil { - return x.Type + return x.Value } return "" } -// Deprecated: Do not use. -func (x *Server) GetName() string { - if x != nil { - return x.Name - } - return "" -} - -func (x *Server) GetDescription() string { - if x != nil { - return x.Description - } - return "" -} - -func (x *Server) GetAddress() string { - if x != nil { - return x.Address - } - return "" -} - -func (x *Server) GetCreateTime() *timestamp.Timestamp { - if x != nil { - return x.CreateTime - } - return nil -} - -func (x *Server) GetUpdateTime() *timestamp.Timestamp { - if x != nil { - return x.UpdateTime - } - return nil -} - -func (x *Server) GetTags() map[string]*TagValues { - if x != nil { - return x.Tags - } - return nil -} - -// TagValues is used because map fields cannot be repeated but can be a -// message -type TagValues struct { +// ServerWorkerStatus is the new message used in place of Server to relay status request info. +type ServerWorkerStatus struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Values []string `protobuf:"bytes,10,rep,name=values,proto3" json:"values,omitempty"` + // Id of the worker. + PublicId string `protobuf:"bytes,10,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty"` + // Name of the resource (optional) + Name string `protobuf:"bytes,20,opt,name=name,proto3" json:"name,omitempty"` + // Address for the worker. This must be set. + Address string `protobuf:"bytes,30,opt,name=address,proto3" json:"address,omitempty"` + // Tags for workers + Tags []*TagPair `protobuf:"bytes,40,rep,name=tags,proto3" json:"tags,omitempty"` } -func (x *TagValues) Reset() { - *x = TagValues{} +func (x *ServerWorkerStatus) Reset() { + *x = ServerWorkerStatus{} if protoimpl.UnsafeEnabled { mi := &file_controller_servers_v1_servers_proto_msgTypes[1] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -160,13 +101,13 @@ func (x *TagValues) Reset() { } } -func (x *TagValues) String() string { +func (x *ServerWorkerStatus) String() string { return protoimpl.X.MessageStringOf(x) } -func (*TagValues) ProtoMessage() {} +func (*ServerWorkerStatus) ProtoMessage() {} -func (x *TagValues) ProtoReflect() protoreflect.Message { +func (x *ServerWorkerStatus) ProtoReflect() protoreflect.Message { mi := &file_controller_servers_v1_servers_proto_msgTypes[1] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -178,14 +119,35 @@ func (x *TagValues) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use TagValues.ProtoReflect.Descriptor instead. -func (*TagValues) Descriptor() ([]byte, []int) { +// Deprecated: Use ServerWorkerStatus.ProtoReflect.Descriptor instead. +func (*ServerWorkerStatus) Descriptor() ([]byte, []int) { return file_controller_servers_v1_servers_proto_rawDescGZIP(), []int{1} } -func (x *TagValues) GetValues() []string { +func (x *ServerWorkerStatus) GetPublicId() string { if x != nil { - return x.Values + return x.PublicId + } + return "" +} + +func (x *ServerWorkerStatus) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *ServerWorkerStatus) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +func (x *ServerWorkerStatus) GetTags() []*TagPair { + if x != nil { + return x.Tags } return nil } @@ -196,45 +158,24 @@ var file_controller_servers_v1_servers_proto_rawDesc = []byte{ 0x0a, 0x23, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2f, 0x76, 0x31, 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x15, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x1a, 0x2f, 0x63, 0x6f, - 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, - 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2f, 0x76, 0x31, 0x2f, 0x74, 0x69, - 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xc1, 0x03, - 0x0a, 0x06, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x69, 0x76, - 0x61, 0x74, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, - 0x69, 0x76, 0x61, 0x74, 0x65, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, - 0x14, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x16, 0x0a, 0x04, 0x6e, - 0x61, 0x6d, 0x65, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x42, 0x02, 0x18, 0x01, 0x52, 0x04, 0x6e, - 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, - 0x6f, 0x6e, 0x18, 0x28, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, - 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, - 0x18, 0x32, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x12, - 0x4b, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x3c, - 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, - 0x61, 0x6d, 0x70, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, - 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4b, 0x0a, 0x0b, - 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x46, 0x20, 0x01, 0x28, - 0x0b, 0x32, 0x2a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, - 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, - 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, - 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x3b, 0x0a, 0x04, 0x74, 0x61, 0x67, - 0x73, 0x18, 0x50, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x27, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, - 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x2e, - 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x2e, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, - 0x52, 0x04, 0x74, 0x61, 0x67, 0x73, 0x1a, 0x59, 0x0a, 0x09, 0x54, 0x61, 0x67, 0x73, 0x45, 0x6e, - 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x36, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0b, 0x32, 0x20, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, - 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x67, - 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, - 0x01, 0x22, 0x23, 0x0a, 0x09, 0x54, 0x61, 0x67, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x12, 0x16, - 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, - 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x42, 0x38, 0x5a, 0x36, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, - 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, - 0x6f, 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, - 0x2f, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, - 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x72, 0x2e, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x22, 0x31, 0x0a, 0x07, + 0x54, 0x61, 0x67, 0x50, 0x61, 0x69, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, + 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x22, + 0x93, 0x01, 0x0a, 0x12, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, + 0x5f, 0x69, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, + 0x63, 0x49, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x14, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, + 0x73, 0x73, 0x18, 0x1e, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, + 0x73, 0x12, 0x32, 0x0a, 0x04, 0x74, 0x61, 0x67, 0x73, 0x18, 0x28, 0x20, 0x03, 0x28, 0x0b, 0x32, + 0x1e, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x65, 0x72, + 0x76, 0x65, 0x72, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x61, 0x67, 0x50, 0x61, 0x69, 0x72, 0x52, + 0x04, 0x74, 0x61, 0x67, 0x73, 0x42, 0x38, 0x5a, 0x36, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, + 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x62, 0x6f, + 0x75, 0x6e, 0x64, 0x61, 0x72, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, + 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x3b, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x62, + 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -249,23 +190,18 @@ func file_controller_servers_v1_servers_proto_rawDescGZIP() []byte { return file_controller_servers_v1_servers_proto_rawDescData } -var file_controller_servers_v1_servers_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_controller_servers_v1_servers_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_controller_servers_v1_servers_proto_goTypes = []interface{}{ - (*Server)(nil), // 0: controller.servers.v1.Server - (*TagValues)(nil), // 1: controller.servers.v1.TagValues - nil, // 2: controller.servers.v1.Server.TagsEntry - (*timestamp.Timestamp)(nil), // 3: controller.storage.timestamp.v1.Timestamp + (*TagPair)(nil), // 0: controller.servers.v1.TagPair + (*ServerWorkerStatus)(nil), // 1: controller.servers.v1.ServerWorkerStatus } var file_controller_servers_v1_servers_proto_depIdxs = []int32{ - 3, // 0: controller.servers.v1.Server.create_time:type_name -> controller.storage.timestamp.v1.Timestamp - 3, // 1: controller.servers.v1.Server.update_time:type_name -> controller.storage.timestamp.v1.Timestamp - 2, // 2: controller.servers.v1.Server.tags:type_name -> controller.servers.v1.Server.TagsEntry - 1, // 3: controller.servers.v1.Server.TagsEntry.value:type_name -> controller.servers.v1.TagValues - 4, // [4:4] is the sub-list for method output_type - 4, // [4:4] is the sub-list for method input_type - 4, // [4:4] is the sub-list for extension type_name - 4, // [4:4] is the sub-list for extension extendee - 0, // [0:4] is the sub-list for field type_name + 0, // 0: controller.servers.v1.ServerWorkerStatus.tags:type_name -> controller.servers.v1.TagPair + 1, // [1:1] is the sub-list for method output_type + 1, // [1:1] is the sub-list for method input_type + 1, // [1:1] is the sub-list for extension type_name + 1, // [1:1] is the sub-list for extension extendee + 0, // [0:1] is the sub-list for field type_name } func init() { file_controller_servers_v1_servers_proto_init() } @@ -275,7 +211,7 @@ func file_controller_servers_v1_servers_proto_init() { } if !protoimpl.UnsafeEnabled { file_controller_servers_v1_servers_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*Server); i { + switch v := v.(*TagPair); i { case 0: return &v.state case 1: @@ -287,7 +223,7 @@ func file_controller_servers_v1_servers_proto_init() { } } file_controller_servers_v1_servers_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { - switch v := v.(*TagValues); i { + switch v := v.(*ServerWorkerStatus); i { case 0: return &v.state case 1: @@ -305,7 +241,7 @@ func file_controller_servers_v1_servers_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_controller_servers_v1_servers_proto_rawDesc, NumEnums: 0, - NumMessages: 3, + NumMessages: 2, NumExtensions: 0, NumServices: 0, }, diff --git a/internal/tests/helper/testing_helper.go b/internal/tests/helper/testing_helper.go index 05bacb66a..24b9cb602 100644 --- a/internal/tests/helper/testing_helper.go +++ b/internal/tests/helper/testing_helper.go @@ -34,7 +34,7 @@ import ( const ( DefaultGracePeriod = time.Second * 15 expectConnectionStateOnControllerTimeout = time.Minute * 2 - expectConnectionStateOnWorkerTimeout = DefaultGracePeriod * 2 + expectConnectionStateOnWorkerTimeout = DefaultGracePeriod * 3 // This is the interval that we check states on in the worker. It // needs to be particularly granular to ensure that we allow for