From 6b48346bf3a1ce285a4bb6a6260fea5c766d66bd Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Thu, 7 Jul 2022 19:02:16 -0400 Subject: [PATCH] Add managed worker address fetching (#2244) --- internal/daemon/worker/status.go | 25 +- .../server_coordination_service.pb.go | 310 +++++++++++++++--- .../server_coordination_service_grpc.pb.go | 38 +++ .../v1/server_coordination_service.proto | 20 ++ 4 files changed, 340 insertions(+), 53 deletions(-) diff --git a/internal/daemon/worker/status.go b/internal/daemon/worker/status.go index fdf59cf80..6b6cb0a90 100644 --- a/internal/daemon/worker/status.go +++ b/internal/daemon/worker/status.go @@ -14,7 +14,9 @@ import ( pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services" "github.com/hashicorp/boundary/internal/observability/event" "github.com/hashicorp/go-secure-stdlib/strutil" + "google.golang.org/grpc/codes" "google.golang.org/grpc/resolver" + "google.golang.org/grpc/status" ) type LastStatusInformation struct { @@ -193,15 +195,36 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess } } else { w.updateTags.Store(false) - // This may be nil if we are in a multiple hop scenario var addrs []resolver.Address var newUpstreams []string + // This may be empty if we are in a multiple hop scenario if len(result.CalculatedUpstreams) > 0 { addrs = make([]resolver.Address, 0, len(result.CalculatedUpstreams)) + newUpstreams = make([]string, 0, len(result.CalculatedUpstreams)) for _, v := range result.CalculatedUpstreams { addrs = append(addrs, resolver.Address{Addr: v.Address}) newUpstreams = append(newUpstreams, v.Address) } + + } else if w.conf.RawConfig.HcpbClusterId != "" { + // This is a worker that is one hop away from managed workers, so attempt to get that list + managedWorkersCtx, managedWorkersCancel := context.WithTimeout(cancelCtx, common.StatusTimeout) + defer managedWorkersCancel() + workersResp, err := client.ManagedWorkers(managedWorkersCtx, &pbs.ManagedWorkersRequest{}) + if err != nil { + if status.Code(err) != codes.Unimplemented { + event.WriteError(managedWorkersCtx, op, err, event.WithInfoMsg("error fetching managed worker information")) + } + } else { + addrs = make([]resolver.Address, 0, len(workersResp.Workers)) + newUpstreams = make([]string, 0, len(workersResp.Workers)) + for _, v := range workersResp.Workers { + addrs = append(addrs, resolver.Address{Addr: v.Address}) + newUpstreams = append(newUpstreams, v.Address) + } + } + } + if len(addrs) > 0 { lastStatus := w.lastStatusSuccess.Load().(*LastStatusInformation) // Compare upstreams; update resolver if there is a difference, and emit an event with old and new addresses if lastStatus != nil && !strutil.EquivalentSlices(lastStatus.LastCalculatedUpstreams, newUpstreams) { 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 81afd84ea..45b8f1d46 100644 --- a/internal/gen/controller/servers/services/server_coordination_service.pb.go +++ b/internal/gen/controller/servers/services/server_coordination_service.pb.go @@ -768,6 +768,151 @@ func (x *StatusResponse) GetWorkerId() string { return "" } +// WorkerInfo contains information about workers for the ManagedWorkerResponse message +type WorkerInfo struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // The id of the worker + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + // The public address of the worker + Address string `protobuf:"bytes,2,opt,name=address,proto3" json:"address,omitempty"` +} + +func (x *WorkerInfo) Reset() { + *x = WorkerInfo{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *WorkerInfo) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*WorkerInfo) ProtoMessage() {} + +func (x *WorkerInfo) ProtoReflect() protoreflect.Message { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[8] + 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 WorkerInfo.ProtoReflect.Descriptor instead. +func (*WorkerInfo) Descriptor() ([]byte, []int) { + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{8} +} + +func (x *WorkerInfo) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *WorkerInfo) GetAddress() string { + if x != nil { + return x.Address + } + return "" +} + +// A managed workers request. Empty for now. +type ManagedWorkersRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *ManagedWorkersRequest) Reset() { + *x = ManagedWorkersRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ManagedWorkersRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ManagedWorkersRequest) ProtoMessage() {} + +func (x *ManagedWorkersRequest) ProtoReflect() protoreflect.Message { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[9] + 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 ManagedWorkersRequest.ProtoReflect.Descriptor instead. +func (*ManagedWorkersRequest) Descriptor() ([]byte, []int) { + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{9} +} + +// A response containing worker information +type ManagedWorkersResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Workers []*WorkerInfo `protobuf:"bytes,1,rep,name=workers,proto3" json:"workers,omitempty"` +} + +func (x *ManagedWorkersResponse) Reset() { + *x = ManagedWorkersResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ManagedWorkersResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ManagedWorkersResponse) ProtoMessage() {} + +func (x *ManagedWorkersResponse) ProtoReflect() protoreflect.Message { + mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[10] + 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 ManagedWorkersResponse.ProtoReflect.Descriptor instead. +func (*ManagedWorkersResponse) Descriptor() ([]byte, []int) { + return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{10} +} + +func (x *ManagedWorkersResponse) GetWorkers() []*WorkerInfo { + if x != nil { + return x.Workers + } + return nil +} + var File_controller_servers_services_v1_server_coordination_service_proto protoreflect.FileDescriptor var file_controller_servers_services_v1_server_coordination_service_proto_rawDesc = []byte{ @@ -864,49 +1009,68 @@ var file_controller_servers_services_v1_server_coordination_service_proto_rawDes 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, + 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x73, 0x22, 0x36, 0x0a, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, + 0x72, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, + 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x22, + 0x17, 0x0a, 0x15, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x5e, 0x0a, 0x16, 0x4d, 0x61, 0x6e, 0x61, + 0x67, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, + 0x73, 0x65, 0x12, 0x44, 0x0a, 0x07, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x18, 0x01, 0x20, + 0x03, 0x28, 0x0b, 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, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x49, 0x6e, 0x66, 0x6f, 0x52, + 0x07, 0x77, 0x6f, 0x72, 0x6b, 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, 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, + 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, 0x8a, + 0x02, 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, 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, + 0x63, 0x65, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, + 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x81, 0x01, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x61, + 0x67, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x12, 0x35, 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, 0x4d, 0x61, 0x6e, 0x61, + 0x67, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x36, 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, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x64, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 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 ( @@ -922,7 +1086,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_rawDe } 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_msgTypes = make([]protoimpl.MessageInfo, 11) 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 @@ -937,7 +1101,10 @@ var file_controller_servers_services_v1_server_coordination_service_proto_goType (*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 + (*WorkerInfo)(nil), // 13: controller.servers.services.v1.WorkerInfo + (*ManagedWorkersRequest)(nil), // 14: controller.servers.services.v1.ManagedWorkersRequest + (*ManagedWorkersResponse)(nil), // 15: controller.servers.services.v1.ManagedWorkersResponse + (*servers.ServerWorkerStatus)(nil), // 16: 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 @@ -948,18 +1115,21 @@ var file_controller_servers_services_v1_server_coordination_service_proto_depIdx 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 + 16, // 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 + 13, // 13: controller.servers.services.v1.ManagedWorkersResponse.workers:type_name -> controller.servers.services.v1.WorkerInfo + 10, // 14: controller.servers.services.v1.ServerCoordinationService.Status:input_type -> controller.servers.services.v1.StatusRequest + 14, // 15: controller.servers.services.v1.ServerCoordinationService.ManagedWorkers:input_type -> controller.servers.services.v1.ManagedWorkersRequest + 12, // 16: controller.servers.services.v1.ServerCoordinationService.Status:output_type -> controller.servers.services.v1.StatusResponse + 15, // 17: controller.servers.services.v1.ServerCoordinationService.ManagedWorkers:output_type -> controller.servers.services.v1.ManagedWorkersResponse + 16, // [16:18] is the sub-list for method output_type + 14, // [14:16] is the sub-list for method input_type + 14, // [14:14] is the sub-list for extension type_name + 14, // [14:14] is the sub-list for extension extendee + 0, // [0:14] is the sub-list for field type_name } func init() { file_controller_servers_services_v1_server_coordination_service_proto_init() } @@ -1064,6 +1234,42 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( return nil } } + file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*WorkerInfo); 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[9].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ManagedWorkersRequest); 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[10].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ManagedWorkersResponse); 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[2].OneofWrappers = []interface{}{ (*Job_SessionInfo)(nil), @@ -1074,7 +1280,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init( GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_controller_servers_services_v1_server_coordination_service_proto_rawDesc, NumEnums: 5, - NumMessages: 8, + NumMessages: 11, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/gen/controller/servers/services/server_coordination_service_grpc.pb.go b/internal/gen/controller/servers/services/server_coordination_service_grpc.pb.go index 58e1a5223..f7d6bce4c 100644 --- a/internal/gen/controller/servers/services/server_coordination_service_grpc.pb.go +++ b/internal/gen/controller/servers/services/server_coordination_service_grpc.pb.go @@ -22,6 +22,8 @@ type ServerCoordinationServiceClient interface { // returns the status response which includes the changes the controller would like to make to // jobs as well as provide a list of the controllers in the system. Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) + // Returns the addresses of managed workers, if any + ManagedWorkers(ctx context.Context, in *ManagedWorkersRequest, opts ...grpc.CallOption) (*ManagedWorkersResponse, error) } type serverCoordinationServiceClient struct { @@ -41,6 +43,15 @@ func (c *serverCoordinationServiceClient) Status(ctx context.Context, in *Status return out, nil } +func (c *serverCoordinationServiceClient) ManagedWorkers(ctx context.Context, in *ManagedWorkersRequest, opts ...grpc.CallOption) (*ManagedWorkersResponse, error) { + out := new(ManagedWorkersResponse) + err := c.cc.Invoke(ctx, "/controller.servers.services.v1.ServerCoordinationService/ManagedWorkers", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + // ServerCoordinationServiceServer is the server API for ServerCoordinationService service. // All implementations must embed UnimplementedServerCoordinationServiceServer // for forward compatibility @@ -49,6 +60,8 @@ type ServerCoordinationServiceServer interface { // returns the status response which includes the changes the controller would like to make to // jobs as well as provide a list of the controllers in the system. Status(context.Context, *StatusRequest) (*StatusResponse, error) + // Returns the addresses of managed workers, if any + ManagedWorkers(context.Context, *ManagedWorkersRequest) (*ManagedWorkersResponse, error) mustEmbedUnimplementedServerCoordinationServiceServer() } @@ -59,6 +72,9 @@ type UnimplementedServerCoordinationServiceServer struct { func (UnimplementedServerCoordinationServiceServer) Status(context.Context, *StatusRequest) (*StatusResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method Status not implemented") } +func (UnimplementedServerCoordinationServiceServer) ManagedWorkers(context.Context, *ManagedWorkersRequest) (*ManagedWorkersResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method ManagedWorkers not implemented") +} func (UnimplementedServerCoordinationServiceServer) mustEmbedUnimplementedServerCoordinationServiceServer() { } @@ -91,6 +107,24 @@ func _ServerCoordinationService_Status_Handler(srv interface{}, ctx context.Cont return interceptor(ctx, in, info, handler) } +func _ServerCoordinationService_ManagedWorkers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ManagedWorkersRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ServerCoordinationServiceServer).ManagedWorkers(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/controller.servers.services.v1.ServerCoordinationService/ManagedWorkers", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ServerCoordinationServiceServer).ManagedWorkers(ctx, req.(*ManagedWorkersRequest)) + } + return interceptor(ctx, in, info, handler) +} + // ServerCoordinationService_ServiceDesc is the grpc.ServiceDesc for ServerCoordinationService service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -102,6 +136,10 @@ var ServerCoordinationService_ServiceDesc = grpc.ServiceDesc{ MethodName: "Status", Handler: _ServerCoordinationService_Status_Handler, }, + { + MethodName: "ManagedWorkers", + Handler: _ServerCoordinationService_ManagedWorkers_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "controller/servers/services/v1/server_coordination_service.proto", 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 ca43cb101..eff5c6998 100644 --- a/internal/proto/controller/servers/services/v1/server_coordination_service.proto +++ b/internal/proto/controller/servers/services/v1/server_coordination_service.proto @@ -11,6 +11,9 @@ service ServerCoordinationService { // returns the status response which includes the changes the controller would like to make to // jobs as well as provide a list of the controllers in the system. rpc Status(StatusRequest) returns (StatusResponse) {} + + // Returns the addresses of managed workers, if any + rpc ManagedWorkers(ManagedWorkersRequest) returns (ManagedWorkersResponse) {} } enum CONNECTIONSTATUS { @@ -122,3 +125,20 @@ message StatusResponse { // controller does not need to do a database lookup for the id using the name field. string worker_id = 40; // @gotags: `class:"public"` } + +// WorkerInfo contains information about workers for the ManagedWorkerResponse message +message WorkerInfo { + // The id of the worker + string id = 1; + + // The public address of the worker + string address = 2; +} + +// A managed workers request. Empty for now. +message ManagedWorkersRequest {} + +// A response containing worker information +message ManagedWorkersResponse { + repeated WorkerInfo workers = 1; +}