Add WithType to worker listing (#2252)

This commit is contained in:
Jeff Mitchell authored and GitHub committed 2022-07-11 15:14:22 -04:00
1 parent 0315c999fe
commit fb3b2209e5
11 files changed
+257 -115

No files matched your search

@@ -22,6 +22,8 @@ import (
"google.golang.org/protobuf/proto"
)
const ManagedWorkerTagKey = "boundary.cloud.hashicorp.com:managed"
type workerServiceServer struct {
pbs.UnimplementedServerCoordinationServiceServer
pbs.UnimplementedSessionServiceServer
@@ -201,6 +203,40 @@ func (ws *workerServiceServer) Status(ctx context.Context, req *pbs.StatusReques
return ret, nil
}
// ListHcpbWorkers looks up workers that are HCP Boundary-managed, currently by
// seeing if they are KMS and have a known tag
func (ws *workerServiceServer) ListHcpbWorkers(ctx context.Context, req *pbs.ListHcpbWorkersRequest) (*pbs.ListHcpbWorkersResponse, error) {
const op = "workers.(workerServiceServer).ListHcpbWorkers"
serversRepo, err := ws.serversRepoFn()
if err != nil {
return nil, status.Errorf(codes.Internal, "Error getting servers repo: %v", err)
}
workers, err := serversRepo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithWorkerType(server.KmsWorkerType))
if err != nil {
return nil, status.Errorf(codes.Internal, "Error looking up workers: %v", err)
}
resp := &pbs.ListHcpbWorkersResponse{}
if len(workers) == 0 {
return resp, nil
}
resp.Workers = make([]*pbs.WorkerInfo, 0, len(workers))
for _, worker := range workers {
vals := worker.CanonicalTags()[ManagedWorkerTagKey]
if len(vals) == 1 && vals[0] == "true" {
resp.Workers = append(resp.Workers, &pbs.WorkerInfo{
Id: worker.GetPublicId(),
Address: worker.GetAddress(),
})
}
}
return resp, nil
}
func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.LookupSessionRequest) (*pbs.LookupSessionResponse, error) {
const op = "workers.(workerServiceServer).LookupSession"
@@ -2,6 +2,7 @@ package handlers_test
import (
"context"
"fmt"
"sync"
"testing"
@@ -174,3 +175,49 @@ func TestLookupSession(t *testing.T) {
})
}
}
// This test creates workers of both kms and pki type and verifies that the only
// returned workers are those of kms type with the expected tag key/value
func TestHcpbWorkers(t *testing.T) {
t.Parallel()
require, assert := require.New(t), assert.New(t)
ctx := context.Background()
conn, _ := db.TestSetup(t, "postgres")
rw := db.New(conn)
wrapper := db.TestWrapper(t)
kms := kms.TestKms(t, conn, wrapper)
serversRepoFn := func() (*server.Repository, error) {
return server.NewRepository(rw, rw, kms)
}
sessionRepoFn := func() (*session.Repository, error) {
return session.NewRepository(rw, rw, kms)
}
connectionRepoFn := func() (*session.ConnectionRepository, error) {
return session.NewConnectionRepository(ctx, rw, rw, kms)
}
for i := 0; i < 3; i++ {
var opt []server.Option
if i > 0 {
// Out of three KMS workers we expect to see two
opt = append(opt, server.WithWorkerTags(&server.Tag{Key: handlers.ManagedWorkerTagKey, Value: "true"}))
}
server.TestKmsWorker(t, conn, wrapper, append(opt, server.WithAddress(fmt.Sprintf("kms.%d", i)))...)
server.TestPkiWorker(t, conn, wrapper, opt...)
}
s := handlers.NewWorkerServiceServer(serversRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms)
require.NotNil(t, s)
res, err := s.ListHcpbWorkers(ctx, &pbs.ListHcpbWorkersRequest{})
require.NoError(err)
require.NotNil(res)
expValues := []string{"kms.1", "kms.2"}
var gotValues []string
for _, worker := range res.Workers {
gotValues = append(gotValues, worker.Address)
}
assert.ElementsMatch(expValues, gotValues)
}
+4 -8
View File
@@ -14,9 +14,7 @@ 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 {
@@ -208,13 +206,11 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context, sessionManager sess
} 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{})
hcpbWorkersCtx, hcpbWorkersCancel := context.WithTimeout(cancelCtx, common.StatusTimeout)
defer hcpbWorkersCancel()
workersResp, err := client.ListHcpbWorkers(hcpbWorkersCtx, &pbs.ListHcpbWorkersRequest{})
if err != nil {
if status.Code(err) != codes.Unimplemented {
event.WriteError(managedWorkersCtx, op, err, event.WithInfoMsg("error fetching managed worker information"))
}
event.WriteError(hcpbWorkersCtx, 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))
@@ -768,7 +768,7 @@ func (x *StatusResponse) GetWorkerId() string {
return ""
}
// WorkerInfo contains information about workers for the ManagedWorkerResponse message
// WorkerInfo contains information about workers for the HcpbWorkerResponse message
type WorkerInfo struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -826,15 +826,15 @@ func (x *WorkerInfo) GetAddress() string {
return ""
}
// A managed workers request. Empty for now.
type ManagedWorkersRequest struct {
// A HCP Boundary workers request. Empty for now.
type ListHcpbWorkersRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
}
func (x *ManagedWorkersRequest) Reset() {
*x = ManagedWorkersRequest{}
func (x *ListHcpbWorkersRequest) Reset() {
*x = ListHcpbWorkersRequest{}
if protoimpl.UnsafeEnabled {
mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -842,13 +842,13 @@ func (x *ManagedWorkersRequest) Reset() {
}
}
func (x *ManagedWorkersRequest) String() string {
func (x *ListHcpbWorkersRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ManagedWorkersRequest) ProtoMessage() {}
func (*ListHcpbWorkersRequest) ProtoMessage() {}
func (x *ManagedWorkersRequest) ProtoReflect() protoreflect.Message {
func (x *ListHcpbWorkersRequest) 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))
@@ -860,13 +860,13 @@ func (x *ManagedWorkersRequest) ProtoReflect() protoreflect.Message {
return mi.MessageOf(x)
}
// Deprecated: Use ManagedWorkersRequest.ProtoReflect.Descriptor instead.
func (*ManagedWorkersRequest) Descriptor() ([]byte, []int) {
// Deprecated: Use ListHcpbWorkersRequest.ProtoReflect.Descriptor instead.
func (*ListHcpbWorkersRequest) Descriptor() ([]byte, []int) {
return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{9}
}
// A response containing worker information
type ManagedWorkersResponse struct {
type ListHcpbWorkersResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
@@ -874,8 +874,8 @@ type ManagedWorkersResponse struct {
Workers []*WorkerInfo `protobuf:"bytes,1,rep,name=workers,proto3" json:"workers,omitempty"`
}
func (x *ManagedWorkersResponse) Reset() {
*x = ManagedWorkersResponse{}
func (x *ListHcpbWorkersResponse) Reset() {
*x = ListHcpbWorkersResponse{}
if protoimpl.UnsafeEnabled {
mi := &file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -883,13 +883,13 @@ func (x *ManagedWorkersResponse) Reset() {
}
}
func (x *ManagedWorkersResponse) String() string {
func (x *ListHcpbWorkersResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ManagedWorkersResponse) ProtoMessage() {}
func (*ListHcpbWorkersResponse) ProtoMessage() {}
func (x *ManagedWorkersResponse) ProtoReflect() protoreflect.Message {
func (x *ListHcpbWorkersResponse) 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))
@@ -901,12 +901,12 @@ func (x *ManagedWorkersResponse) ProtoReflect() protoreflect.Message {
return mi.MessageOf(x)
}
// Deprecated: Use ManagedWorkersResponse.ProtoReflect.Descriptor instead.
func (*ManagedWorkersResponse) Descriptor() ([]byte, []int) {
// Deprecated: Use ListHcpbWorkersResponse.ProtoReflect.Descriptor instead.
func (*ListHcpbWorkersResponse) Descriptor() ([]byte, []int) {
return file_controller_servers_services_v1_server_coordination_service_proto_rawDescGZIP(), []int{10}
}
func (x *ManagedWorkersResponse) GetWorkers() []*WorkerInfo {
func (x *ListHcpbWorkersResponse) GetWorkers() []*WorkerInfo {
if x != nil {
return x.Workers
}
@@ -1013,64 +1013,64 @@ var file_controller_servers_services_v1_server_coordination_service_proto_rawDes
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, 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,
0x18, 0x0a, 0x16, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65,
0x72, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x5f, 0x0a, 0x17, 0x4c, 0x69, 0x73,
0x74, 0x48, 0x63, 0x70, 0x62, 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, 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,
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, 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, 0x8d, 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, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x84, 0x01, 0x0a, 0x0f, 0x4c, 0x69,
0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x12, 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, 0x4c,
0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x37, 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, 0x4c, 0x69, 0x73, 0x74, 0x48, 0x63, 0x70, 0x62, 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 (
@@ -1102,8 +1102,8 @@ var file_controller_servers_services_v1_server_coordination_service_proto_goType
(*JobChangeRequest)(nil), // 11: controller.servers.services.v1.JobChangeRequest
(*StatusResponse)(nil), // 12: controller.servers.services.v1.StatusResponse
(*WorkerInfo)(nil), // 13: controller.servers.services.v1.WorkerInfo
(*ManagedWorkersRequest)(nil), // 14: controller.servers.services.v1.ManagedWorkersRequest
(*ManagedWorkersResponse)(nil), // 15: controller.servers.services.v1.ManagedWorkersResponse
(*ListHcpbWorkersRequest)(nil), // 14: controller.servers.services.v1.ListHcpbWorkersRequest
(*ListHcpbWorkersResponse)(nil), // 15: controller.servers.services.v1.ListHcpbWorkersResponse
(*servers.ServerWorkerStatus)(nil), // 16: controller.servers.v1.ServerWorkerStatus
}
var file_controller_servers_services_v1_server_coordination_service_proto_depIdxs = []int32{
@@ -1120,11 +1120,11 @@ var file_controller_servers_services_v1_server_coordination_service_proto_depIdx
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
13, // 13: controller.servers.services.v1.ManagedWorkersResponse.workers:type_name -> controller.servers.services.v1.WorkerInfo
13, // 13: controller.servers.services.v1.ListHcpbWorkersResponse.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
14, // 15: controller.servers.services.v1.ServerCoordinationService.ListHcpbWorkers:input_type -> controller.servers.services.v1.ListHcpbWorkersRequest
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
15, // 17: controller.servers.services.v1.ServerCoordinationService.ListHcpbWorkers:output_type -> controller.servers.services.v1.ListHcpbWorkersResponse
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
@@ -1247,7 +1247,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init(
}
}
file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[9].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*ManagedWorkersRequest); i {
switch v := v.(*ListHcpbWorkersRequest); i {
case 0:
return &v.state
case 1:
@@ -1259,7 +1259,7 @@ func file_controller_servers_services_v1_server_coordination_service_proto_init(
}
}
file_controller_servers_services_v1_server_coordination_service_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*ManagedWorkersResponse); i {
switch v := v.(*ListHcpbWorkersResponse); i {
case 0:
return &v.state
case 1:
@@ -22,8 +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)
// Returns the addresses of HCP Boundary workers, if any
ListHcpbWorkers(ctx context.Context, in *ListHcpbWorkersRequest, opts ...grpc.CallOption) (*ListHcpbWorkersResponse, error)
}
type serverCoordinationServiceClient struct {
@@ -43,9 +43,9 @@ 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...)
func (c *serverCoordinationServiceClient) ListHcpbWorkers(ctx context.Context, in *ListHcpbWorkersRequest, opts ...grpc.CallOption) (*ListHcpbWorkersResponse, error) {
out := new(ListHcpbWorkersResponse)
err := c.cc.Invoke(ctx, "/controller.servers.services.v1.ServerCoordinationService/ListHcpbWorkers", in, out, opts...)
if err != nil {
return nil, err
}
@@ -60,8 +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)
// Returns the addresses of HCP Boundary workers, if any
ListHcpbWorkers(context.Context, *ListHcpbWorkersRequest) (*ListHcpbWorkersResponse, error)
mustEmbedUnimplementedServerCoordinationServiceServer()
}
@@ -72,8 +72,8 @@ 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) ListHcpbWorkers(context.Context, *ListHcpbWorkersRequest) (*ListHcpbWorkersResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ListHcpbWorkers not implemented")
}
func (UnimplementedServerCoordinationServiceServer) mustEmbedUnimplementedServerCoordinationServiceServer() {
}
@@ -107,20 +107,20 @@ 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)
func _ServerCoordinationService_ListHcpbWorkers_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListHcpbWorkersRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ServerCoordinationServiceServer).ManagedWorkers(ctx, in)
return srv.(ServerCoordinationServiceServer).ListHcpbWorkers(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/controller.servers.services.v1.ServerCoordinationService/ManagedWorkers",
FullMethod: "/controller.servers.services.v1.ServerCoordinationService/ListHcpbWorkers",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ServerCoordinationServiceServer).ManagedWorkers(ctx, req.(*ManagedWorkersRequest))
return srv.(ServerCoordinationServiceServer).ListHcpbWorkers(ctx, req.(*ListHcpbWorkersRequest))
}
return interceptor(ctx, in, info, handler)
}
@@ -137,8 +137,8 @@ var ServerCoordinationService_ServiceDesc = grpc.ServiceDesc{
Handler: _ServerCoordinationService_Status_Handler,
},
{
MethodName: "ManagedWorkers",
Handler: _ServerCoordinationService_ManagedWorkers_Handler,
MethodName: "ListHcpbWorkers",
Handler: _ServerCoordinationService_ListHcpbWorkers_Handler,
},
},
Streams: []grpc.StreamDesc{},
@@ -12,8 +12,8 @@ service ServerCoordinationService {
// 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) {}
// Returns the addresses of HCP Boundary workers, if any
rpc ListHcpbWorkers(ListHcpbWorkersRequest) returns (ListHcpbWorkersResponse) {}
}
enum CONNECTIONSTATUS {
@@ -126,7 +126,7 @@ message StatusResponse {
string worker_id = 40; // @gotags: `class:"public"`
}
// WorkerInfo contains information about workers for the ManagedWorkerResponse message
// WorkerInfo contains information about workers for the HcpbWorkerResponse message
message WorkerInfo {
// The id of the worker
string id = 1;
@@ -135,10 +135,10 @@ message WorkerInfo {
string address = 2;
}
// A managed workers request. Empty for now.
message ManagedWorkersRequest {}
// A HCP Boundary workers request. Empty for now.
message ListHcpbWorkersRequest {}
// A response containing worker information
message ManagedWorkersResponse {
message ListHcpbWorkersResponse {
repeated WorkerInfo workers = 1;
}
+9
View File
@@ -38,6 +38,7 @@ type options struct {
withFetchNodeCredentialsRequest *types.FetchNodeCredentialsRequest
withTestPkiWorkerAuthorized bool
withTestPkiWorkerKeyId *string
withWorkerType WorkerType
}
func getDefaultOptions() options {
@@ -163,3 +164,11 @@ func WithTestPkiWorkerAuthorizedKeyId(id *string) Option {
o.withTestPkiWorkerKeyId = id
}
}
// WithWorkerType allows specifying a particular type of worker (kms, pki)
// during lookup or listing
func WithWorkerType(with WorkerType) Option {
return func(o *options) {
o.withWorkerType = with
}
}
+6
View File
@@ -158,4 +158,10 @@ func Test_GetOpts(t *testing.T) {
opts.withNewIdFunc = nil
assert.Equal(opts, testOpts)
})
t.Run("WithWorkerType", func(t *testing.T) {
opts := getDefaultOptions()
assert.Empty(t, opts.withWorkerType)
opts = getOpts(WithWorkerType(KmsWorkerType))
assert.Equal(t, KmsWorkerType, opts.withWorkerType)
})
}
+10
View File
@@ -162,6 +162,7 @@ func lookupWorker(ctx context.Context, reader db.Reader, id string) (*Worker, er
// then the last status update time is ignored.
// If WithLimit < 0, then unlimited results are returned. If WithLimit == 0, then
// default limits are used for results.
// Also supports: WithWorkerType
func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...Option) ([]*Worker, error) {
const op = "server.(Repository).ListWorkers"
switch {
@@ -185,6 +186,15 @@ func (r *Repository) ListWorkers(ctx context.Context, scopeIds []string, opt ...
whereArgs = append(whereArgs, scopeIds)
}
switch opts.withWorkerType {
case "":
case KmsWorkerType, PkiWorkerType:
where = append(where, "type = ?")
whereArgs = append(whereArgs, opts.withWorkerType.String())
default:
return nil, errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("unknown worker type %v", opts.withWorkerType))
}
limit := r.defaultLimit
if opts.withLimit != 0 {
// non-zero signals an override of the default limit for the repo.
+38 -1
View File
@@ -515,9 +515,32 @@ func TestListWorkers(t *testing.T) {
createCnt: testLimit + 1,
reqScopes: []string{scope.Global.String()},
opts: []server.Option{server.WithLimit(-1)},
wantCnt: testLimit*2 + 2,
wantErr: false,
},
{
name: "only-kms-type",
createCnt: testLimit + 1,
reqScopes: []string{scope.Global.String()},
opts: []server.Option{server.WithLimit(-1), server.WithWorkerType(server.KmsWorkerType)},
wantCnt: testLimit + 1,
wantErr: false,
},
{
name: "only-pki-type",
createCnt: testLimit + 1,
reqScopes: []string{scope.Global.String()},
opts: []server.Option{server.WithLimit(-1), server.WithWorkerType(server.PkiWorkerType)},
wantCnt: testLimit + 1,
wantErr: false,
},
{
name: "bad-type",
createCnt: testLimit + 1,
reqScopes: []string{scope.Global.String()},
opts: []server.Option{server.WithLimit(-1), server.WithWorkerType("foo")},
wantErr: true,
},
{
name: "default-limit",
createCnt: testLimit + 1,
@@ -545,7 +568,10 @@ func TestListWorkers(t *testing.T) {
w := server.NewWorker(scope.Global.String())
db.TestDeleteWhere(t, conn, w, "true")
for i := 0; i < tt.createCnt; i++ {
server.TestKmsWorker(t, conn, wrapper)
server.TestKmsWorker(t, conn, wrapper, server.WithWorkerTags(&server.Tag{"key", "val"}))
}
for i := 0; i < tt.createCnt; i++ {
server.TestPkiWorker(t, conn, wrapper)
}
// the purpose of these tests isn't to check liveness, so disable
// liveness checking.
@@ -557,6 +583,17 @@ func TestListWorkers(t *testing.T) {
}
require.NoError(t, err)
assert.Len(t, got, tt.wantCnt)
for _, worker := range got {
switch worker.GetType() {
case server.KmsWorkerType.String():
canonicalTags := worker.CanonicalTags()
assert.Len(t, canonicalTags, 1)
require.Len(t, canonicalTags["key"], 1)
assert.Equal(t, canonicalTags["key"][0], "val")
case server.PkiWorkerType.String():
assert.Empty(t, worker.CanonicalTags())
}
}
})
}
}
+2 -1
View File
@@ -126,7 +126,8 @@ func (w *Worker) ActiveConnectionCount() uint32 {
}
// CanonicalTags is the deduplicated set of tags contained on both the resource
// set over the API as well as the tags reported by the worker itself.
// set over the API as well as the tags reported by the worker itself. This
// function is guaranteed to return a non-nil map.
func (w *Worker) CanonicalTags() map[string][]string {
dedupedTags := make(map[Tag]struct{})
for _, t := range w.apiTags {