From 91e23eaeb42782b4ea6bf4aab897884a11365706 Mon Sep 17 00:00:00 2001 From: orion Date: Thu, 29 Oct 2026 12:03:49 +0000 Subject: [PATCH 1/2] feat(vermeer): expose cluster health snapshots Report administrator-scoped Master and Worker readiness, collect process CPU and RSS independently, and cover stream lifecycle and availability checks. --- vermeer/README.md | 34 ++ vermeer/apps/common/os_util.go | 3 + vermeer/apps/common/process_metrics.go | 105 +++++ vermeer/apps/common/process_metrics_test.go | 92 +++++ vermeer/apps/master/services/http_cluster.go | 205 +++++++++ .../apps/master/services/http_cluster_test.go | 155 +++++++ vermeer/apps/master/services/http_master.go | 2 +- vermeer/apps/master/services/router.go | 1 + vermeer/apps/master/services/service.go | 21 + vermeer/apps/master/workers/worker_health.go | 112 +++++ .../apps/master/workers/worker_health_test.go | 252 +++++++++++ vermeer/apps/protos/worker.pb.go | 390 +++++++++++------- vermeer/apps/protos/worker.proto | 10 +- vermeer/apps/worker/grpc_handlers.go | 15 +- vermeer/apps/worker/process_metrics_test.go | 62 +++ 15 files changed, 1294 insertions(+), 165 deletions(-) create mode 100644 vermeer/apps/common/process_metrics.go create mode 100644 vermeer/apps/common/process_metrics_test.go create mode 100644 vermeer/apps/master/services/http_cluster.go create mode 100644 vermeer/apps/master/services/http_cluster_test.go create mode 100644 vermeer/apps/master/workers/worker_health.go create mode 100644 vermeer/apps/master/workers/worker_health_test.go create mode 100644 vermeer/apps/worker/process_metrics_test.go diff --git a/vermeer/README.md b/vermeer/README.md index a2abb2754..646f84fac 100644 --- a/vermeer/README.md +++ b/vermeer/README.md @@ -690,6 +690,40 @@ vermeer/tools/protoc/linux64/protoc vermeer/apps/protos/*.proto --go-grpc_out=ve ## Monitoring +### Cluster health snapshot + +`GET /api/v1/admin/cluster` returns a read-only, versioned snapshot for Hubble's +operations overview. In token-authenticated deployments it requires an admin +token; it uses the same authentication filters as the other `/admin` APIs. + +The response includes `schema_version: 1`, `observed_at` (Unix milliseconds), +Master and Worker details, `registered_workers`, and `ready_workers`. +Registration alone is not readiness: each Worker needs a ready reverse gRPC +connection and live load, compute, and superstep streams. No Workers, or a +registered Worker missing a readiness check, yields `DEGRADED` with a reason. +Removed Workers are not fabricated as offline nodes, and the registered count +is not a Kubernetes desired replica count. + +Each node reports resource collection separately in `metric_status`. +`metrics.cpu_cores` is consumed process CPU seconds per elapsed wall second; +`metrics.rss_bytes` is resident process memory, not host or Pod utilization. +The first sample can report `metrics_warming_up`; subsequent observations +include `sample_window_seconds`. Resource collection failure never changes +Worker readiness, and unsupported older Workers return an explicit unsupported +metric status. Resource RPCs share a bounded deadline and run outside the Worker +manager lock. + +`GET /api/v1/workers/availability` implements the accompanying Helm availability +hook's space-scoped check. It returns only `{"available": true|false}` for the +authenticated credential's space, following the existing allocation +rules: an accessible common, space, or graph Worker group must contain at least +one Worker and every registered member must pass all readiness checks. A Worker +registered in another group's scope does not make this check succeed. This +endpoint describes space task availability, rather than overall cluster health. + +Build and deploy matching Vermeer and Hubble images together to use the new +cluster-health contract. + Access the Web UI dashboard at `http://master-ip:6688/` for: > **Note**: The Web UI requires `auth=token` in `master.ini` (or `-auth token` via command line). Without token authentication, the UI routes are not registered. diff --git a/vermeer/apps/common/os_util.go b/vermeer/apps/common/os_util.go index 64ff4c84d..e99d95609 100644 --- a/vermeer/apps/common/os_util.go +++ b/vermeer/apps/common/os_util.go @@ -51,6 +51,9 @@ func IsFileOrDirExist(dir string) bool { func MachineMemUsedPercent() (float64, error) { v, err := mem.VirtualMemory() + if err != nil { + return 0, err + } return v.UsedPercent, err } diff --git a/vermeer/apps/common/process_metrics.go b/vermeer/apps/common/process_metrics.go new file mode 100644 index 000000000..336b0efc5 --- /dev/null +++ b/vermeer/apps/common/process_metrics.go @@ -0,0 +1,105 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package common + +import ( + "errors" + "math" + "os" + "sync" + "time" + + "github.com/shirou/gopsutil/v3/process" +) + +// ProcessMetrics describes this process, rather than the host or a Kubernetes Pod. +// CPUCores is CPU seconds consumed divided by elapsed wall time; it may exceed 1. +type ProcessMetrics struct { + CPUCores float64 + RSSBytes uint64 + SampleWindowSeconds float64 + ObservedAt int64 +} + +var errMetricsWarmingUp = errors.New("metrics_warming_up") +var errProcessMetricsUnavailable = errors.New("process_metrics_unavailable") + +type processMetricsReader func() (cpuSeconds float64, rssBytes uint64, err error) + +type processMetricsSampler struct { + mu sync.Mutex + read processMetricsReader + now func() time.Time + lastAttempt time.Time + baselineAt time.Time + baselineCPU float64 + result ProcessMetrics + err error +} + +var currentProcessMetrics = processMetricsSampler{read: readCurrentProcessMetrics, now: time.Now} + +// ReadProcessMetrics samples without waiting for a measurement interval. Calls +// within one second reuse the last observation. The first call establishes a baseline +// and returns metrics_warming_up instead of inventing a zero CPU measurement. +func ReadProcessMetrics() (ProcessMetrics, error) { + return currentProcessMetrics.sample() +} + +func (s *processMetricsSampler) sample() (ProcessMetrics, error) { + s.mu.Lock() + defer s.mu.Unlock() + now := s.now() + if !s.lastAttempt.IsZero() && now.Sub(s.lastAttempt) >= 0 && now.Sub(s.lastAttempt) < time.Second { + return s.result, s.err + } + s.lastAttempt = now + s.result = ProcessMetrics{ObservedAt: now.UnixMilli()} + cpuSeconds, rssBytes, err := s.read() + if err != nil || math.IsNaN(cpuSeconds) || math.IsInf(cpuSeconds, 0) || cpuSeconds < 0 { + s.err = errProcessMetricsUnavailable + return s.result, s.err + } + s.result.RSSBytes = rssBytes + if s.baselineAt.IsZero() || !now.After(s.baselineAt) || cpuSeconds < s.baselineCPU { + s.baselineAt, s.baselineCPU = now, cpuSeconds + s.err = errMetricsWarmingUp + return s.result, s.err + } + s.result.SampleWindowSeconds = now.Sub(s.baselineAt).Seconds() + s.result.CPUCores = (cpuSeconds - s.baselineCPU) / s.result.SampleWindowSeconds + s.baselineAt, s.baselineCPU = now, cpuSeconds + s.err = nil + return s.result, nil +} + +func readCurrentProcessMetrics() (float64, uint64, error) { + p, err := process.NewProcess(int32(os.Getpid())) + if err != nil { + return 0, 0, err + } + times, err := p.Times() + if err != nil { + return 0, 0, err + } + memory, err := p.MemoryInfo() + if err != nil { + return 0, 0, err + } + return times.User + times.System, memory.RSS, nil +} diff --git a/vermeer/apps/common/process_metrics_test.go b/vermeer/apps/common/process_metrics_test.go new file mode 100644 index 000000000..26ad6b63f --- /dev/null +++ b/vermeer/apps/common/process_metrics_test.go @@ -0,0 +1,92 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package common + +import ( + "errors" + "math" + "sync" + "testing" + "time" +) + +func TestProcessMetricsMeasureCoresAndCacheObservation(t *testing.T) { + now := time.Unix(100, 0) + cpuSeconds := 10.0 + reads := 0 + s := processMetricsSampler{ + now: func() time.Time { return now }, + read: func() (float64, uint64, error) { + reads++ + return cpuSeconds, 12345, nil + }, + } + if _, err := s.sample(); !errors.Is(err, errMetricsWarmingUp) { + t.Fatalf("first observation must warm up, got %v", err) + } + now = now.Add(2 * time.Second) + cpuSeconds += 3 + got, err := s.sample() + if err != nil || got.CPUCores != 1.5 || got.RSSBytes != 12345 || got.SampleWindowSeconds != 2 { + t.Fatalf("expected 1.5 process CPU cores with 2s window, got %+v, %v", got, err) + } + now = now.Add(100 * time.Millisecond) + cached, err := s.sample() + if err != nil || cached != got || reads != 2 { + t.Fatalf("cached observation changed: %+v, reads=%d, err=%v", cached, reads, err) + } +} + +func TestProcessMetricsFailureDoesNotFabricateUsage(t *testing.T) { + now := time.Unix(100, 0) + invalid := false + s := processMetricsSampler{ + now: func() time.Time { return now }, + read: func() (float64, uint64, error) { + if invalid { + return math.NaN(), 100, nil + } + return 5, 100, nil + }, + } + _, _ = s.sample() + now = now.Add(time.Second) + invalid = true + got, err := s.sample() + if !errors.Is(err, errProcessMetricsUnavailable) || got.RSSBytes != 0 || got.SampleWindowSeconds != 0 { + t.Fatalf("invalid sample must be unavailable, got %+v, %v", got, err) + } +} + +func TestProcessMetricsConcurrentReadsShareBaseline(t *testing.T) { + s := processMetricsSampler{ + now: func() time.Time { return time.Unix(100, 0) }, + read: func() (float64, uint64, error) { return 5, 100, nil }, + } + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := s.sample(); !errors.Is(err, errMetricsWarmingUp) { + t.Errorf("concurrent initial sample: %v", err) + } + }() + } + wg.Wait() +} diff --git a/vermeer/apps/master/services/http_cluster.go b/vermeer/apps/master/services/http_cluster.go new file mode 100644 index 000000000..f930e5f8a --- /dev/null +++ b/vermeer/apps/master/services/http_cluster.go @@ -0,0 +1,205 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package services + +import ( + "context" + "math" + "net/http" + "sync" + "time" + "vermeer/apps/common" + "vermeer/apps/master/workers" + pb "vermeer/apps/protos" + + "github.com/gin-gonic/gin" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type ClusterMetricStatus struct { + Availability string `json:"availability"` + ObservedAt int64 `json:"observed_at"` + Reason string `json:"reason,omitempty"` +} + +type ClusterNode struct { + ID string `json:"id"` + Name string `json:"name"` + Role string `json:"role"` + Group string `json:"group,omitempty"` + GrpcPeer string `json:"grpc_peer"` + Version string `json:"version"` + LaunchTime time.Time `json:"launch_time"` + Status string `json:"status"` + Ready bool `json:"ready"` + Reason string `json:"reason,omitempty"` + Checks map[string]bool `json:"checks"` + Metrics map[string]interface{} `json:"metrics"` + MetricStatus ClusterMetricStatus `json:"metric_status"` +} + +type ClusterSnapshot struct { + SchemaVersion int `json:"schema_version"` + ObservedAt int64 `json:"observed_at"` + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + RegisteredWorkers int `json:"registered_workers"` + ReadyWorkers int `json:"ready_workers"` + Master ClusterNode `json:"master"` + Workers []ClusterNode `json:"workers"` +} + +// Registered exclusively in SetAdminRouters, after the configured auth/admin +// filters. Like the existing admin APIs, anonymous mode retains its semantics. +type AdminClusterHandler struct{ common.SenHandler } + +func (h *AdminClusterHandler) GET(ctx *gin.Context) { + ctx.JSON(http.StatusOK, collectClusterSnapshot(ctx.Request.Context())) +} + +func collectClusterSnapshot(parent context.Context) ClusterSnapshot { + now := time.Now().UnixMilli() + snapshots := WorkerMgr.HealthSnapshots() + info := InfoMaster.snapshot() + result := ClusterSnapshot{ + SchemaVersion: 1, ObservedAt: now, Status: "UP", + RegisteredWorkers: len(snapshots), + Master: clusterMasterNode(info, now), + Workers: make([]ClusterNode, len(snapshots)), + } + // One total deadline and bounded fanout prevent telemetry from delaying health + // indefinitely. Worker metadata is already copied, so no lock spans an RPC. + ctx, cancel := context.WithTimeout(parent, 800*time.Millisecond) + defer cancel() + var wg sync.WaitGroup + jobs := make(chan int) + count := len(snapshots) + if count > 8 { + count = 8 + } + for i := 0; i < count; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for index := range jobs { + result.Workers[index] = clusterWorkerNode(ctx, snapshots[index], now) + } + }() + } + for index := range snapshots { + jobs <- index + } + close(jobs) + wg.Wait() + for _, worker := range result.Workers { + if worker.Ready { + result.ReadyWorkers++ + } + } + switch { + case !result.Master.Ready: + result.Status, result.Reason = "DEGRADED", result.Master.Reason + case len(snapshots) == 0: + result.Status, result.Reason = "DEGRADED", "no_workers" + case result.ReadyWorkers != result.RegisteredWorkers: + result.Status, result.Reason = "DEGRADED", "workers_not_ready" + } + return result +} + +func clusterMasterNode(info MasterInfoSnapshot, now int64) ClusterNode { + offline := common.GetConfigDefault("service_offline", false) + checks := map[string]bool{ + "http_ready": true, + "initialized": !info.LaunchTime.IsZero(), + "service_online": offline != true && offline != "true", + } + ready := checks["initialized"] && checks["service_online"] + node := ClusterNode{ID: "master", Name: "Vermeer Master", Role: "MASTER", + GrpcPeer: info.GrpcPeer, Version: info.Version, LaunchTime: info.LaunchTime, + Status: "UP", Ready: ready, Checks: checks, Metrics: map[string]interface{}{}} + if !ready { + node.Status, node.Reason = "DEGRADED", "master_initializing" + if !checks["service_online"] { + node.Reason = "service_offline" + } + } + metrics, err := common.ReadProcessMetrics() + if err != nil { + node.MetricStatus = ClusterMetricStatus{"UNAVAILABLE", now, err.Error()} + } else { + node.Metrics = processMetricValues(metrics.CPUCores, metrics.RSSBytes, metrics.SampleWindowSeconds) + node.MetricStatus = ClusterMetricStatus{"AVAILABLE", metrics.ObservedAt, ""} + } + return node +} + +func clusterWorkerNode(ctx context.Context, snapshot workers.WorkerHealthSnapshot, now int64) ClusterNode { + checks := snapshot.Checks + node := ClusterNode{ID: snapshot.Name, Name: snapshot.Name, Role: "WORKER", + Group: snapshot.Group, GrpcPeer: snapshot.GrpcPeer, Version: snapshot.Version, + LaunchTime: snapshot.LaunchTime, Status: "UP", Ready: checks.Ready(), + Reason: checks.Reason(), Checks: map[string]bool{ + "registered": checks.Registered, "connection_ready": checks.ConnectionReady, + "load_stream_ready": checks.LoadStreamReady, "compute_stream_ready": checks.ComputeStreamReady, + "superstep_stream_ready": checks.SuperstepStreamReady, + }, Metrics: map[string]interface{}{}, + MetricStatus: ClusterMetricStatus{"UNAVAILABLE", now, "connection_not_ready"}, + } + if !node.Ready { + node.Status = "DEGRADED" + } + if !checks.ConnectionReady || snapshot.Session == nil { + return node + } + response, err := snapshot.Session.GetWorkerStatInfo(ctx, &pb.WorkerStatInfoReq{}) + if err != nil { + reason := "metrics_rpc_failed" + availability := "UNAVAILABLE" + switch status.Code(err) { + case codes.Unimplemented: + availability, reason = "UNSUPPORTED", "runtime_metrics_unsupported" + case codes.DeadlineExceeded, codes.Canceled: + reason = "metrics_timeout" + } + node.MetricStatus = ClusterMetricStatus{availability, now, reason} + return node + } + if response == nil || response.GetMetricSchemaVersion() != 1 { + node.MetricStatus = ClusterMetricStatus{"UNSUPPORTED", now, "runtime_metrics_unsupported"} + return node + } + if response.GetMetricError() != "" { + node.MetricStatus = ClusterMetricStatus{"UNAVAILABLE", now, response.GetMetricError()} + return node + } + cpu, window := response.GetProcessCpuCores(), response.GetSampleWindowSeconds() + if cpu < 0 || math.IsNaN(cpu) || math.IsInf(cpu, 0) || window <= 0 || + math.IsNaN(window) || math.IsInf(window, 0) || response.GetMetricsObservedAt() <= 0 { + node.MetricStatus = ClusterMetricStatus{"UNAVAILABLE", now, "invalid_runtime_metrics"} + return node + } + node.Metrics = processMetricValues(cpu, response.GetProcessRssBytes(), window) + node.MetricStatus = ClusterMetricStatus{"AVAILABLE", response.GetMetricsObservedAt(), ""} + return node +} + +func processMetricValues(cpu float64, rss uint64, window float64) map[string]interface{} { + return map[string]interface{}{"cpu_cores": cpu, "rss_bytes": rss, "sample_window_seconds": window} +} diff --git a/vermeer/apps/master/services/http_cluster_test.go b/vermeer/apps/master/services/http_cluster_test.go new file mode 100644 index 000000000..0dfde6406 --- /dev/null +++ b/vermeer/apps/master/services/http_cluster_test.go @@ -0,0 +1,155 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package services + +import ( + "context" + "encoding/json" + "errors" + "github.com/gin-gonic/gin" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "math" + "net/http/httptest" + "testing" + "time" + "vermeer/apps/auth" + "vermeer/apps/common" + "vermeer/apps/master/workers" + pb "vermeer/apps/protos" + "vermeer/apps/structure" +) + +type clusterStatsClient struct { + pb.WorkerClient + response *pb.WorkerStatInfoResp + err error +} + +func (client clusterStatsClient) GetWorkerStatInfo(context.Context, *pb.WorkerStatInfoReq, ...grpc.CallOption) (*pb.WorkerStatInfoResp, error) { + return client.response, client.err +} + +func readyWorkerSnapshot(client pb.WorkerClient) workers.WorkerHealthSnapshot { + return workers.WorkerHealthSnapshot{Name: "worker-one", Group: "group-one", Session: client, + Checks: workers.ReadinessChecks{Registered: true, ConnectionReady: true, + LoadStreamReady: true, ComputeStreamReady: true, SuperstepStreamReady: true}} +} + +func TestClusterWorkerMetricsNeverOverrideReadiness(t *testing.T) { + for _, test := range []struct { + name string + response *pb.WorkerStatInfoResp + err error + availability string + }{ + {"legacy", &pb.WorkerStatInfoResp{}, nil, "UNSUPPORTED"}, + {"unimplemented", nil, status.Error(codes.Unimplemented, "old worker"), "UNSUPPORTED"}, + {"timeout", nil, status.Error(codes.DeadlineExceeded, "timeout"), "UNAVAILABLE"}, + {"failed", nil, errors.New("internal failure"), "UNAVAILABLE"}, + {"warming", &pb.WorkerStatInfoResp{MetricSchemaVersion: 1, MetricError: "metrics_warming_up"}, nil, "UNAVAILABLE"}, + {"invalid", &pb.WorkerStatInfoResp{MetricSchemaVersion: 1, ProcessCpuCores: math.NaN()}, nil, "UNAVAILABLE"}, + } { + t.Run(test.name, func(t *testing.T) { + node := clusterWorkerNode(context.Background(), readyWorkerSnapshot(clusterStatsClient{response: test.response, err: test.err}), 1234) + if !node.Ready || node.Status != "UP" || node.Reason != "" { + t.Fatalf("metrics changed readiness: %+v", node) + } + if node.MetricStatus.Availability != test.availability || len(node.Metrics) != 0 { + t.Fatalf("bad metrics failure: %+v", node) + } + }) + } + snapshot := readyWorkerSnapshot(clusterStatsClient{response: &pb.WorkerStatInfoResp{MetricSchemaVersion: 1, + ProcessCpuCores: 1.5, ProcessRssBytes: 1024, SampleWindowSeconds: 2, MetricsObservedAt: 1000}}) + node := clusterWorkerNode(context.Background(), snapshot, 1234) + if node.Metrics["cpu_cores"] != 1.5 || node.Metrics["rss_bytes"] != uint64(1024) || node.MetricStatus.ObservedAt != 1000 { + t.Fatalf("wrong process metric units: %+v", node) + } + snapshot.Checks.ComputeStreamReady = false + node = clusterWorkerNode(context.Background(), snapshot, 1234) + if node.Ready || node.Status != "DEGRADED" || node.Reason != "compute_stream_not_ready" || node.MetricStatus.Availability != "AVAILABLE" { + t.Fatalf("stream failure conflated with metrics: %+v", node) + } +} + +func TestClusterEmptySnapshotAndMaster(t *testing.T) { + previous := InfoMaster + InfoMaster = &Info{LaunchTime: time.Now(), Version: "test", GrpcPeer: "master:6689"} + t.Cleanup(func() { InfoMaster = previous }) + common.SetConfig("service_offline", false) + result := collectClusterSnapshot(context.Background()) + if result.SchemaVersion != 1 || result.ObservedAt <= 0 || result.Status != "DEGRADED" || result.Reason != "no_workers" || + result.RegisteredWorkers != 0 || result.ReadyWorkers != 0 || result.Workers == nil || !result.Master.Ready { + t.Fatalf("bad empty snapshot: %+v", result) + } + common.SetConfig("service_offline", true) + t.Cleanup(func() { common.SetConfig("service_offline", false) }) + node := clusterMasterNode(InfoMaster.snapshot(), 1234) + if node.Ready || node.Reason != "service_offline" { + t.Fatalf("offline master ready: %+v", node) + } +} + +func TestClusterRouteRequiresAdminAuthentication(t *testing.T) { + gin.SetMode(gin.TestMode) + common.SetConfig("auth_token_factor", "cluster-test-only-factor") + auth.Init() + sentinel := &common.Sentinel{} + sentinel.Init("127.0.0.1:0", "release") + SetAdminRouters(sentinel, auth.TokenFilter, auth.AdminFilter) + factory := &auth.TokenFactory{Factor: "cluster-test-only-factor"} + tokenFor := func(user string) string { + token := factory.NewToken(user, structure.AdminSpace, "cluster-test") + factory.Sign(token) + encoded, _ := json.Marshal(token) + return auth.ToBase58(string(encoded)) + } + for _, test := range []struct { + name, token string + expected int + }{ + {"anonymous", "", 401}, {"member", tokenFor("member"), 401}, {"admin", tokenFor("admin"), 200}, + } { + t.Run(test.name, func(t *testing.T) { + req := httptest.NewRequest("GET", "/api/v1/admin/cluster", nil) + if test.token != "" { + req.Header.Set("Authorization", "Bearer "+test.token) + } + recorder := httptest.NewRecorder() + sentinel.ServeHTTP(recorder, req) + if recorder.Code != test.expected { + t.Fatalf("status %d, want %d", recorder.Code, test.expected) + } + if test.expected == 200 { + var body map[string]interface{} + if err := json.Unmarshal(recorder.Body.Bytes(), &body); err != nil || body["schema_version"] != float64(1) { + t.Fatalf("bad snapshot: %s", recorder.Body.String()) + } + } + }) + } + req := httptest.NewRequest("POST", "/api/v1/admin/cluster", nil) + req.Header.Set("Authorization", "Bearer "+tokenFor("admin")) + recorder := httptest.NewRecorder() + sentinel.ServeHTTP(recorder, req) + if recorder.Code < 400 { + t.Fatal("cluster endpoint accepted mutation") + } +} diff --git a/vermeer/apps/master/services/http_master.go b/vermeer/apps/master/services/http_master.go index d46dace80..8d37e90c7 100644 --- a/vermeer/apps/master/services/http_master.go +++ b/vermeer/apps/master/services/http_master.go @@ -30,6 +30,6 @@ type InfoMasterHandler struct { func (mh *InfoMasterHandler) GET(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{ - "master": InfoMaster, + "master": InfoMaster.snapshot(), }) } diff --git a/vermeer/apps/master/services/router.go b/vermeer/apps/master/services/router.go index 7474d1138..1661b5343 100644 --- a/vermeer/apps/master/services/router.go +++ b/vermeer/apps/master/services/router.go @@ -77,6 +77,7 @@ func SetRouters(sen *common.Sentinel, authFilters ...gin.HandlerFunc) { func SetAdminRouters(sen *common.Sentinel, authFilters ...gin.HandlerFunc) { // /admin regVerAPI(sen, 1, "/admin", map[string]common.BaseHandler{ + "/cluster": &AdminClusterHandler{}, "/graphs/:space_name": &AdminGraphsHandler{}, "/graphs/:space_name/create": &AdminGraphCreateHandler{}, "/graphs/:space_name/:graph_name": &AdminGraphHandler{}, diff --git a/vermeer/apps/master/services/service.go b/vermeer/apps/master/services/service.go index e64d4ac03..2842bcbe3 100644 --- a/vermeer/apps/master/services/service.go +++ b/vermeer/apps/master/services/service.go @@ -18,8 +18,10 @@ under the License. package services import ( + "sync" "time" "vermeer/algorithms" + "vermeer/apps/common" "vermeer/apps/compute" "vermeer/apps/graphio" "vermeer/apps/master/access" @@ -49,6 +51,7 @@ var ServiceMaster = &service{} var InfoMaster = &Info{} type Info struct { + mu sync.RWMutex GrpcPeer string `json:"grpc_peer,omitempty"` IpAddr string `json:"ip_addr,omitempty"` DebugMod string `json:"debug_mod,omitempty"` @@ -56,6 +59,21 @@ type Info struct { LaunchTime time.Time `json:"launch_time,omitempty"` } +type MasterInfoSnapshot struct { + GrpcPeer string `json:"grpc_peer,omitempty"` + IpAddr string `json:"ip_addr,omitempty"` + DebugMod string `json:"debug_mod,omitempty"` + Version string `json:"version,omitempty"` + LaunchTime time.Time `json:"launch_time,omitempty"` +} + +func (info *Info) snapshot() MasterInfoSnapshot { + info.mu.RLock() + defer info.mu.RUnlock() + return MasterInfoSnapshot{GrpcPeer: info.GrpcPeer, IpAddr: info.IpAddr, + DebugMod: info.DebugMod, Version: info.Version, LaunchTime: info.LaunchTime} +} + type service struct { //LoadGraphHandler *LoadGraphServer //ComputerHandler *ComputeTaskServer @@ -155,6 +173,9 @@ func (s *service) Init() error { } func (s *service) Run() { + _, _ = common.ReadProcessMetrics() + InfoMaster.mu.Lock() + defer InfoMaster.mu.Unlock() InfoMaster.LaunchTime = time.Now() } diff --git a/vermeer/apps/master/workers/worker_health.go b/vermeer/apps/master/workers/worker_health.go new file mode 100644 index 000000000..1e767965e --- /dev/null +++ b/vermeer/apps/master/workers/worker_health.go @@ -0,0 +1,112 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package workers + +import ( + "sort" + "time" + pb "vermeer/apps/protos" + + "google.golang.org/grpc/connectivity" +) + +// WorkerHealthSnapshot is a value copy. Live sessions are retained only for a +// bounded metrics RPC after releasing the manager lock, never serialized. +type WorkerHealthSnapshot struct { + ID int32 + Name string + Group string + GrpcPeer string + Version string + LaunchTime time.Time + Checks ReadinessChecks + Session pb.WorkerClient +} + +type ReadinessChecks struct { + Registered bool `json:"registered"` + ConnectionReady bool `json:"connection_ready"` + LoadStreamReady bool `json:"load_stream_ready"` + ComputeStreamReady bool `json:"compute_stream_ready"` + SuperstepStreamReady bool `json:"superstep_stream_ready"` + sessionEnded bool +} + +func (checks ReadinessChecks) Ready() bool { + return !checks.sessionEnded && checks.Registered && checks.ConnectionReady && checks.LoadStreamReady && + checks.ComputeStreamReady && checks.SuperstepStreamReady +} + +func (checks ReadinessChecks) Reason() string { + switch { + case !checks.Registered: + return "worker_not_registered" + case !checks.ConnectionReady: + return "connection_not_ready" + case !checks.LoadStreamReady: + return "load_stream_not_ready" + case !checks.ComputeStreamReady: + return "compute_stream_not_ready" + case !checks.SuperstepStreamReady: + return "superstep_stream_not_ready" + case checks.sessionEnded: + return "worker_session_ended" + default: + return "" + } +} + +func (wm *workerManager) HealthSnapshots() []WorkerHealthSnapshot { + defer wm.Unlock(wm.Lock()) + result := make([]WorkerHealthSnapshot, 0, len(wm.workersByName)) + for _, worker := range wm.workersByName { + checks := wm.readinessChecksLocked(worker) + snapshot := WorkerHealthSnapshot{ + ID: worker.Id, Name: worker.Name, Group: worker.Group, + GrpcPeer: worker.GrpcPeer, Version: worker.Version, + LaunchTime: worker.LaunchTime, Checks: checks, + } + // Registration writes Session before publishing Connection under this lock. + // Do not read that field while a registration is still in progress. + if checks.ConnectionReady { + snapshot.Session = worker.Session + } + result = append(result, snapshot) + } + sort.Slice(result, func(i, j int) bool { return result[i].ID < result[j].ID }) + return result +} + +func (wm *workerManager) readinessChecksLocked(worker *WorkerClient) ReadinessChecks { + checks := ReadinessChecks{} + if worker == nil || wm.workersByName[worker.Name] != worker { + return checks + } + checks.Registered = true + checks.ConnectionReady = worker.Connection != nil && worker.Connection.GetState() == connectivity.Ready + live := func(stream TaskStream) bool { + ctx := worker.taskStreams[stream] + return ctx != nil && ctx.Err() == nil + } + checks.LoadStreamReady = live(LoadTaskStream) + checks.ComputeStreamReady = live(ComputeTaskStream) + checks.SuperstepStreamReady = live(SuperStepStream) + // Reuse the authoritative lifecycle gate, including offline-cleanup ownership. + checks.sessionEnded = checks.Ready() && !wm.workerReadyLocked(worker) + return checks +} diff --git a/vermeer/apps/master/workers/worker_health_test.go b/vermeer/apps/master/workers/worker_health_test.go new file mode 100644 index 000000000..c888c5c26 --- /dev/null +++ b/vermeer/apps/master/workers/worker_health_test.go @@ -0,0 +1,252 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package workers + +import ( + "context" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" + "net" + "sync" + "testing" + "time" + pb "vermeer/apps/protos" +) + +func testReadyConnection(t *testing.T) *grpc.ClientConn { + t.Helper() + listener := bufconn.Listen(1024 * 1024) + server := grpc.NewServer() + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + t.Cleanup(func() { _ = listener.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + connection, err := grpc.DialContext(ctx, "passthrough:///worker", grpc.WithContextDialer( + func(context.Context, string) (net.Conn, error) { return listener.Dial() }), + grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = connection.Close() }) + return connection +} + +func TestHealthSnapshotsRequireConnectionAndAllTaskStreams(t *testing.T) { + worker := &WorkerClient{Id: 1, Name: "worker-one", Group: "group-one"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + first := manager.HealthSnapshots() + if first[0].Checks.Ready() || first[0].Checks.Reason() != "connection_not_ready" { + t.Fatalf("registration must not imply readiness: %+v", first[0]) + } + manager.SetWorkerConnection(worker, testReadyConnection(t)) + for stream := LoadTaskStream; stream < taskStreamCount; stream++ { + if manager.HealthSnapshots()[0].Checks.Ready() { + t.Fatal("missing task stream reported ready") + } + manager.SetTaskStream(worker, stream, context.Background()) + } + ready := manager.HealthSnapshots()[0] + if !ready.Checks.Ready() || ready.Checks.Reason() != "" { + t.Fatalf("not ready: %+v", ready) + } + ctx, cancel := context.WithCancel(context.Background()) + manager.SetTaskStream(worker, ComputeTaskStream, ctx) + cancel() + if check := manager.HealthSnapshots()[0].Checks; check.Ready() || check.Reason() != "compute_stream_not_ready" { + t.Fatalf("canceled stream remained ready: %+v", check) + } + manager.Lock() + worker.Group = "changed" + manager.Unlock(nil) + if ready.Group != "group-one" { + t.Fatal("snapshot aliases mutable metadata") + } + manager.Lock() + delete(manager.workersByName, worker.Name) + manager.Unlock(nil) + if remaining := manager.HealthSnapshots(); len(remaining) != 0 { + t.Fatal("removed worker was fabricated") + } +} + +func TestOldTaskStreamCannotClearReplacement(t *testing.T) { + worker := &WorkerClient{Name: "worker-one"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + old, cancelOld := context.WithCancel(context.Background()) + defer cancelOld() + current, cancelCurrent := context.WithCancel(context.Background()) + defer cancelCurrent() + manager.SetTaskStream(worker, LoadTaskStream, old) + manager.SetTaskStream(worker, LoadTaskStream, current) + manager.ClearTaskStream(worker, LoadTaskStream, old) + if !manager.HealthSnapshots()[0].Checks.LoadStreamReady { + t.Fatal("old stream cleared replacement") + } + replacement := &WorkerClient{Name: worker.Name} + manager.workersByName[worker.Name] = replacement + manager.SetTaskStream(replacement, LoadTaskStream, current) + manager.ClearTaskStream(worker, LoadTaskStream, current) + manager.SetTaskStream(worker, LoadTaskStream, old) + if replacement.taskStreams[LoadTaskStream] != current { + t.Fatal("old worker modified replacement session") + } +} + +func TestConcurrentStreamChangesAndHealthSnapshots(t *testing.T) { + worker := &WorkerClient{Name: "worker-one"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + var wg sync.WaitGroup + for stream := LoadTaskStream; stream < taskStreamCount; stream++ { + wg.Add(1) + go func(stream TaskStream) { + defer wg.Done() + for i := 0; i < 100; i++ { + ctx, cancel := context.WithCancel(context.Background()) + manager.SetTaskStream(worker, stream, ctx) + _ = manager.HealthSnapshots() + cancel() + manager.ClearTaskStream(worker, stream, ctx) + } + }(stream) + } + wg.Wait() +} + +func TestWorkerAvailabilityRespectsGroupsAndCompleteReadiness(t *testing.T) { + connection := testReadyConnection(t) + for _, test := range []struct { + name string + groups map[string]string + graphNames []string + workerGroups []string + incomplete int + want bool + }{ + {name: "empty", incomplete: -1}, + {name: "common ready group", workerGroups: []string{"$"}, incomplete: -1, want: true}, + {name: "foreign group", workerGroups: []string{"foreign"}, incomplete: -1}, + {name: "explicit space group", groups: map[string]string{"own-space": "dedicated"}, + workerGroups: []string{"dedicated"}, incomplete: -1, want: true}, + {name: "explicit group blocks common fallback", groups: map[string]string{"own-space": "missing"}, + workerGroups: []string{"$"}, incomplete: -1}, + {name: "own graph group", groups: map[string]string{"own-space graph": "dedicated"}, + graphNames: []string{"graph"}, workerGroups: []string{"dedicated"}, incomplete: -1, want: true}, + {name: "another space graph does not leak", groups: map[string]string{"other-space graph": "dedicated"}, + graphNames: []string{"graph"}, workerGroups: []string{"dedicated"}, incomplete: -1}, + {name: "one incomplete member fails entire group", workerGroups: []string{"$", "$"}, incomplete: 1}, + {name: "ready own graph group survives incomplete common group", groups: map[string]string{"own-space graph": "dedicated"}, + graphNames: []string{"graph"}, workerGroups: []string{"$", "dedicated"}, incomplete: 0, want: true}, + {name: "ready foreign group cannot rescue own incomplete group", groups: map[string]string{"own-space": "own"}, + workerGroups: []string{"own", "foreign"}, incomplete: 0}, + } { + t.Run(test.name, func(t *testing.T) { + manager := &workerManager{workersByName: map[string]*WorkerClient{}, groupMapper: test.groups} + for index, group := range test.workerGroups { + worker := &WorkerClient{Name: group + string(rune('a'+index)), Group: group} + manager.workersByName[worker.Name] = worker + manager.SetWorkerConnection(worker, connection) + for stream := LoadTaskStream; stream < taskStreamCount; stream++ { + if index == test.incomplete && stream == ComputeTaskStream { + continue + } + manager.SetTaskStream(worker, stream, context.Background()) + } + } + if got := manager.HasAvailableWorkers("own-space", test.graphNames); got != test.want { + t.Fatalf("availability %v, want %v", got, test.want) + } + }) + } +} + +func TestWorkerAvailabilityCannotUsePendingOrCanceledConnections(t *testing.T) { + worker := &WorkerClient{Name: "worker", Group: "$"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + for stream := LoadTaskStream; stream < taskStreamCount; stream++ { + manager.SetTaskStream(worker, stream, context.Background()) + } + if manager.HasAvailableWorkers("space", nil) { + t.Fatal("streams without reverse connection reported available") + } + connection := testReadyConnection(t) + manager.SetWorkerConnection(worker, connection) + if !manager.HasAvailableWorkers("space", nil) { + t.Fatal("ready common worker unavailable") + } + ctx, cancel := context.WithCancel(context.Background()) + manager.SetTaskStream(worker, SuperStepStream, ctx) + cancel() + if manager.HasAvailableWorkers("space", nil) { + t.Fatal("canceled task stream reported available") + } + manager.SetTaskStream(worker, SuperStepStream, context.Background()) + _ = connection.Close() + if manager.HasAvailableWorkers("space", nil) { + t.Fatal("closed reverse connection reported available") + } +} + +func TestHealthSnapshotHonorsUpstreamOfflineCleanup(t *testing.T) { + worker := &WorkerClient{Name: "worker", Group: "$"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + manager.SetWorkerConnection(worker, testReadyConnection(t)) + for stream := LoadTaskStream; stream < taskStreamCount; stream++ { + manager.SetTaskStream(worker, stream, context.Background()) + } + manager.Lock() + worker.offlineCleanup = true + manager.Unlock(nil) + snapshot := manager.HealthSnapshots()[0] + if snapshot.Checks.Ready() || snapshot.Checks.Reason() != "worker_session_ended" || + manager.IsWorkerReady(worker) || manager.HasAvailableWorkers("space", nil) { + t.Fatalf("snapshot overrode upstream cleanup: %+v", snapshot.Checks) + } +} + +func TestHealthSnapshotWaitsForSessionPublication(t *testing.T) { + worker := &WorkerClient{Name: "worker", Group: "$"} + manager := &workerManager{workersByName: map[string]*WorkerClient{worker.Name: worker}} + connection := testReadyConnection(t) + started, published := make(chan struct{}), make(chan struct{}) + var readers sync.WaitGroup + readers.Add(1) + go func() { + defer readers.Done() + close(started) + for { + _ = manager.HealthSnapshots() + select { + case <-published: + return + default: + } + } + }() + <-started + // Mirror upstream registration: Session is assigned before Connection is + // published under the manager lock. Snapshot readers must not race that write. + worker.Session = pb.NewWorkerClient(connection) + manager.SetWorkerConnection(worker, connection) + close(published) + readers.Wait() + if manager.HealthSnapshots()[0].Session == nil { + t.Fatal("published worker session was omitted") + } +} diff --git a/vermeer/apps/protos/worker.pb.go b/vermeer/apps/protos/worker.pb.go index ab6eb93c2..8ed67a0fd 100644 --- a/vermeer/apps/protos/worker.pb.go +++ b/vermeer/apps/protos/worker.pb.go @@ -1617,6 +1617,14 @@ type WorkerStatInfoResp struct { Base *BaseResponse `protobuf:"bytes,1,opt,name=Base,proto3" json:"Base,omitempty"` MemMachineUsedPercent float64 `protobuf:"fixed64,2,opt,name=MemMachineUsedPercent,proto3" json:"MemMachineUsedPercent,omitempty"` MemProcessUsedPercent float64 `protobuf:"fixed64,3,opt,name=MemProcessUsedPercent,proto3" json:"MemProcessUsedPercent,omitempty"` + // Process resource snapshots. CPU is consumed CPU seconds / elapsed seconds, + // measured in cores, and RSS is absolute resident bytes, not host/Pod ratios. + MetricSchemaVersion uint32 `protobuf:"varint,4,opt,name=metric_schema_version,json=metricSchemaVersion,proto3" json:"metric_schema_version,omitempty"` + ProcessCpuCores float64 `protobuf:"fixed64,5,opt,name=process_cpu_cores,json=processCpuCores,proto3" json:"process_cpu_cores,omitempty"` + ProcessRssBytes uint64 `protobuf:"varint,6,opt,name=process_rss_bytes,json=processRssBytes,proto3" json:"process_rss_bytes,omitempty"` + SampleWindowSeconds float64 `protobuf:"fixed64,7,opt,name=sample_window_seconds,json=sampleWindowSeconds,proto3" json:"sample_window_seconds,omitempty"` + MetricsObservedAt int64 `protobuf:"varint,8,opt,name=metrics_observed_at,json=metricsObservedAt,proto3" json:"metrics_observed_at,omitempty"` + MetricError string `protobuf:"bytes,9,opt,name=metric_error,json=metricError,proto3" json:"metric_error,omitempty"` } func (x *WorkerStatInfoResp) Reset() { @@ -1672,6 +1680,48 @@ func (x *WorkerStatInfoResp) GetMemProcessUsedPercent() float64 { return 0 } +func (x *WorkerStatInfoResp) GetMetricSchemaVersion() uint32 { + if x != nil { + return x.MetricSchemaVersion + } + return 0 +} + +func (x *WorkerStatInfoResp) GetProcessCpuCores() float64 { + if x != nil { + return x.ProcessCpuCores + } + return 0 +} + +func (x *WorkerStatInfoResp) GetProcessRssBytes() uint64 { + if x != nil { + return x.ProcessRssBytes + } + return 0 +} + +func (x *WorkerStatInfoResp) GetSampleWindowSeconds() float64 { + if x != nil { + return x.SampleWindowSeconds + } + return 0 +} + +func (x *WorkerStatInfoResp) GetMetricsObservedAt() int64 { + if x != nil { + return x.MetricsObservedAt + } + return 0 +} + +func (x *WorkerStatInfoResp) GetMetricError() string { + if x != nil { + return x.MetricError + } + return "" +} + type SettingActionReq struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -1884,6 +1934,7 @@ type RuntimeActionReq struct { unknownFields protoimpl.UnknownFields // Types that are assignable to Request: + // // *RuntimeActionReq_HostInfoReq // *RuntimeActionReq_MemoryLimitReq // *RuntimeActionReq_CPULimitReq @@ -2151,6 +2202,7 @@ type RuntimeActionResp struct { unknownFields protoimpl.UnknownFields // Types that are assignable to Response: + // // *RuntimeActionResp_HostInfoResp // *RuntimeActionResp_MemoryLimitResp // *RuntimeActionResp_CPULimitResp @@ -2600,7 +2652,7 @@ var file_worker_proto_rawDesc = []byte{ 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, 0x42, - 0x61, 0x73, 0x65, 0x22, 0xaa, 0x01, 0x0a, 0x12, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, + 0x61, 0x73, 0x65, 0x22, 0xbd, 0x03, 0x0a, 0x12, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x52, 0x04, @@ -2611,173 +2663,191 @@ var file_worker_proto_rawDesc = []byte{ 0x6d, 0x50, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x55, 0x73, 0x65, 0x64, 0x50, 0x65, 0x72, 0x63, 0x65, 0x6e, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x01, 0x52, 0x15, 0x4d, 0x65, 0x6d, 0x50, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x55, 0x73, 0x65, 0x64, 0x50, 0x65, 0x72, 0x63, 0x65, 0x6e, 0x74, - 0x22, 0xf2, 0x01, 0x0a, 0x10, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, - 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, - 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, - 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x12, 0x2d, - 0x0a, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, - 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, - 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, - 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x54, - 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x50, 0x61, 0x72, 0x74, 0x49, 0x64, 0x18, - 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x50, 0x61, 0x72, 0x74, 0x49, 0x64, 0x12, 0x1e, 0x0a, - 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x10, 0x0a, - 0x03, 0x45, 0x6e, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08, 0x52, 0x03, 0x45, 0x6e, 0x64, 0x12, - 0x10, 0x0a, 0x03, 0x4e, 0x75, 0x6d, 0x18, 0x07, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x4e, 0x75, - 0x6d, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0c, 0x52, - 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0xf3, 0x01, 0x0a, 0x11, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, - 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x12, 0x27, 0x0a, 0x04, 0x42, - 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, - 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, - 0x42, 0x61, 0x73, 0x65, 0x12, 0x2d, 0x0a, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, - 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x06, 0x61, 0x63, 0x74, - 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x18, 0x03, 0x20, - 0x01, 0x28, 0x05, 0x52, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x50, - 0x61, 0x72, 0x74, 0x49, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x50, 0x61, 0x72, - 0x74, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, 0x61, 0x6d, - 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, - 0x61, 0x6d, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x45, 0x6e, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08, - 0x52, 0x03, 0x45, 0x6e, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x4e, 0x75, 0x6d, 0x18, 0x07, 0x20, 0x01, - 0x28, 0x05, 0x52, 0x03, 0x4e, 0x75, 0x6d, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, - 0x08, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0xda, 0x01, 0x0a, 0x10, - 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, - 0x12, 0x3a, 0x0a, 0x0b, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x18, - 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, - 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x48, 0x00, 0x52, - 0x0b, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12, 0x43, 0x0a, 0x0e, - 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, - 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x48, - 0x00, 0x52, 0x0e, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, - 0x71, 0x12, 0x3a, 0x0a, 0x0b, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, - 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, - 0x53, 0x65, 0x74, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x48, 0x00, - 0x52, 0x0b, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x42, 0x09, 0x0a, - 0x07, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x39, 0x0a, 0x0e, 0x47, 0x65, 0x74, 0x48, - 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, - 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, - 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, 0x42, - 0x61, 0x73, 0x65, 0x22, 0xc0, 0x01, 0x0a, 0x11, 0x53, 0x65, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, - 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, 0x73, - 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, - 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, 0x42, 0x61, - 0x73, 0x65, 0x12, 0x24, 0x0a, 0x0d, 0x4d, 0x61, 0x78, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x55, - 0x73, 0x65, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x4d, 0x61, 0x78, 0x4d, 0x65, - 0x6d, 0x6f, 0x72, 0x79, 0x55, 0x73, 0x65, 0x64, 0x12, 0x28, 0x0a, 0x0f, 0x4d, 0x69, 0x6e, 0x52, - 0x65, 0x6d, 0x61, 0x69, 0x6e, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, - 0x0d, 0x52, 0x0f, 0x4d, 0x69, 0x6e, 0x52, 0x65, 0x6d, 0x61, 0x69, 0x6e, 0x4d, 0x65, 0x6d, 0x6f, - 0x72, 0x79, 0x12, 0x32, 0x0a, 0x14, 0x53, 0x6f, 0x66, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, - 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x61, 0x74, 0x69, 0x6f, 0x18, 0x04, 0x20, 0x01, 0x28, 0x02, - 0x52, 0x14, 0x53, 0x6f, 0x66, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, - 0x74, 0x52, 0x61, 0x74, 0x69, 0x6f, 0x22, 0x51, 0x0a, 0x0e, 0x53, 0x65, 0x74, 0x43, 0x50, 0x55, - 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, + 0x12, 0x32, 0x0a, 0x15, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x5f, 0x73, 0x63, 0x68, 0x65, 0x6d, + 0x61, 0x5f, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, 0x52, + 0x13, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x56, 0x65, 0x72, + 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x2a, 0x0a, 0x11, 0x70, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x5f, + 0x63, 0x70, 0x75, 0x5f, 0x63, 0x6f, 0x72, 0x65, 0x73, 0x18, 0x05, 0x20, 0x01, 0x28, 0x01, 0x52, + 0x0f, 0x70, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x43, 0x70, 0x75, 0x43, 0x6f, 0x72, 0x65, 0x73, + 0x12, 0x2a, 0x0a, 0x11, 0x70, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x72, 0x73, 0x73, 0x5f, + 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x06, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0f, 0x70, 0x72, 0x6f, + 0x63, 0x65, 0x73, 0x73, 0x52, 0x73, 0x73, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x32, 0x0a, 0x15, + 0x73, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x5f, 0x77, 0x69, 0x6e, 0x64, 0x6f, 0x77, 0x5f, 0x73, 0x65, + 0x63, 0x6f, 0x6e, 0x64, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x01, 0x52, 0x13, 0x73, 0x61, 0x6d, + 0x70, 0x6c, 0x65, 0x57, 0x69, 0x6e, 0x64, 0x6f, 0x77, 0x53, 0x65, 0x63, 0x6f, 0x6e, 0x64, 0x73, + 0x12, 0x2e, 0x0a, 0x13, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x5f, 0x6f, 0x62, 0x73, 0x65, + 0x72, 0x76, 0x65, 0x64, 0x5f, 0x61, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x03, 0x52, 0x11, 0x6d, + 0x65, 0x74, 0x72, 0x69, 0x63, 0x73, 0x4f, 0x62, 0x73, 0x65, 0x72, 0x76, 0x65, 0x64, 0x41, 0x74, + 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x5f, 0x65, 0x72, 0x72, 0x6f, 0x72, + 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x45, 0x72, + 0x72, 0x6f, 0x72, 0x22, 0xf2, 0x01, 0x0a, 0x10, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, + 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, 0x42, 0x61, 0x73, - 0x65, 0x12, 0x16, 0x0a, 0x06, 0x4d, 0x61, 0x78, 0x43, 0x50, 0x55, 0x18, 0x02, 0x20, 0x01, 0x28, - 0x0d, 0x52, 0x06, 0x4d, 0x61, 0x78, 0x43, 0x50, 0x55, 0x22, 0xe5, 0x01, 0x0a, 0x11, 0x52, 0x75, - 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x12, - 0x3d, 0x0a, 0x0c, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x18, - 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, - 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x48, 0x00, - 0x52, 0x0c, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x12, 0x46, - 0x0a, 0x0f, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, - 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, + 0x65, 0x12, 0x2d, 0x0a, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, + 0x0e, 0x32, 0x15, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, + 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, + 0x12, 0x16, 0x0a, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, + 0x52, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x50, 0x61, 0x72, 0x74, + 0x49, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x50, 0x61, 0x72, 0x74, 0x49, 0x64, + 0x12, 0x1e, 0x0a, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x4e, 0x61, 0x6d, 0x65, + 0x12, 0x10, 0x0a, 0x03, 0x45, 0x6e, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x08, 0x52, 0x03, 0x45, + 0x6e, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x4e, 0x75, 0x6d, 0x18, 0x07, 0x20, 0x01, 0x28, 0x05, 0x52, + 0x03, 0x4e, 0x75, 0x6d, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x08, 0x20, 0x01, + 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0xf3, 0x01, 0x0a, 0x11, 0x53, 0x65, 0x74, + 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x12, 0x27, + 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, + 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x12, 0x2d, 0x0a, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, + 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, + 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x06, + 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x16, 0x0a, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, + 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, 0x54, 0x61, 0x73, 0x6b, 0x49, 0x64, 0x12, 0x16, + 0x0a, 0x06, 0x50, 0x61, 0x72, 0x74, 0x49, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x06, + 0x50, 0x61, 0x72, 0x74, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x57, 0x6f, 0x72, 0x6b, + 0x65, 0x72, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x45, 0x6e, 0x64, 0x18, 0x06, 0x20, + 0x01, 0x28, 0x08, 0x52, 0x03, 0x45, 0x6e, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x4e, 0x75, 0x6d, 0x18, + 0x07, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x4e, 0x75, 0x6d, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, + 0x74, 0x61, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0xda, + 0x01, 0x0a, 0x10, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, + 0x52, 0x65, 0x71, 0x12, 0x3a, 0x0a, 0x0b, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, + 0x65, 0x71, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x47, 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, + 0x48, 0x00, 0x52, 0x0b, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12, + 0x43, 0x0a, 0x0e, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, + 0x71, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, - 0x65, 0x73, 0x70, 0x48, 0x00, 0x52, 0x0f, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, - 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x12, 0x3d, 0x0a, 0x0c, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, - 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, - 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, - 0x74, 0x52, 0x65, 0x73, 0x70, 0x48, 0x00, 0x52, 0x0c, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, - 0x74, 0x52, 0x65, 0x73, 0x70, 0x42, 0x0a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, - 0x65, 0x22, 0xa3, 0x01, 0x0a, 0x0f, 0x47, 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, - 0x6f, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, - 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, - 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x12, - 0x20, 0x0a, 0x0b, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0b, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x4d, 0x65, 0x6d, 0x6f, 0x72, - 0x79, 0x12, 0x28, 0x0a, 0x0f, 0x41, 0x76, 0x61, 0x69, 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x4d, 0x65, - 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0f, 0x41, 0x76, 0x61, 0x69, - 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x12, 0x1a, 0x0a, 0x08, 0x43, - 0x50, 0x55, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x43, - 0x50, 0x55, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x3e, 0x0a, 0x12, 0x53, 0x65, 0x74, 0x4d, 0x65, - 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, + 0x65, 0x71, 0x48, 0x00, 0x52, 0x0e, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, + 0x74, 0x52, 0x65, 0x71, 0x12, 0x3a, 0x0a, 0x0b, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, + 0x52, 0x65, 0x71, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, + 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, + 0x71, 0x48, 0x00, 0x52, 0x0b, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, + 0x42, 0x09, 0x0a, 0x07, 0x72, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x39, 0x0a, 0x0e, 0x47, + 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, + 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, + 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x22, 0xc0, 0x01, 0x0a, 0x11, 0x53, 0x65, 0x74, 0x4d, 0x65, + 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, + 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, + 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, + 0x04, 0x42, 0x61, 0x73, 0x65, 0x12, 0x24, 0x0a, 0x0d, 0x4d, 0x61, 0x78, 0x4d, 0x65, 0x6d, 0x6f, + 0x72, 0x79, 0x55, 0x73, 0x65, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0d, 0x4d, 0x61, + 0x78, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x55, 0x73, 0x65, 0x64, 0x12, 0x28, 0x0a, 0x0f, 0x4d, + 0x69, 0x6e, 0x52, 0x65, 0x6d, 0x61, 0x69, 0x6e, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0f, 0x4d, 0x69, 0x6e, 0x52, 0x65, 0x6d, 0x61, 0x69, 0x6e, 0x4d, + 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x12, 0x32, 0x0a, 0x14, 0x53, 0x6f, 0x66, 0x74, 0x4d, 0x65, 0x6d, + 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x61, 0x74, 0x69, 0x6f, 0x18, 0x04, 0x20, + 0x01, 0x28, 0x02, 0x52, 0x14, 0x53, 0x6f, 0x66, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, + 0x69, 0x6d, 0x69, 0x74, 0x52, 0x61, 0x74, 0x69, 0x6f, 0x22, 0x51, 0x0a, 0x0e, 0x53, 0x65, 0x74, + 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x71, 0x12, 0x27, 0x0a, 0x04, 0x42, + 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, + 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x52, 0x04, + 0x42, 0x61, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x4d, 0x61, 0x78, 0x43, 0x50, 0x55, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x0d, 0x52, 0x06, 0x4d, 0x61, 0x78, 0x43, 0x50, 0x55, 0x22, 0xe5, 0x01, 0x0a, + 0x11, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, + 0x73, 0x70, 0x12, 0x3d, 0x0a, 0x0c, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, + 0x73, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x47, 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, + 0x70, 0x48, 0x00, 0x52, 0x0c, 0x48, 0x6f, 0x73, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, + 0x70, 0x12, 0x46, 0x0a, 0x0f, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, + 0x52, 0x65, 0x73, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x6d, 0x61, 0x73, + 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, + 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x48, 0x00, 0x52, 0x0f, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, + 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x12, 0x3d, 0x0a, 0x0c, 0x43, 0x50, 0x55, + 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, + 0x17, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x43, 0x50, 0x55, 0x4c, + 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x48, 0x00, 0x52, 0x0c, 0x43, 0x50, 0x55, 0x4c, + 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x42, 0x0a, 0x0a, 0x08, 0x72, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x22, 0xa3, 0x01, 0x0a, 0x0f, 0x47, 0x65, 0x74, 0x48, 0x6f, 0x73, 0x74, + 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, + 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x52, 0x04, 0x42, 0x61, + 0x73, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x4d, 0x65, 0x6d, 0x6f, 0x72, + 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0b, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x4d, 0x65, + 0x6d, 0x6f, 0x72, 0x79, 0x12, 0x28, 0x0a, 0x0f, 0x41, 0x76, 0x61, 0x69, 0x6c, 0x61, 0x62, 0x6c, + 0x65, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0f, 0x41, + 0x76, 0x61, 0x69, 0x6c, 0x61, 0x62, 0x6c, 0x65, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x12, 0x1a, + 0x0a, 0x08, 0x43, 0x50, 0x55, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0d, + 0x52, 0x08, 0x43, 0x50, 0x55, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x22, 0x3e, 0x0a, 0x12, 0x53, 0x65, + 0x74, 0x4d, 0x65, 0x6d, 0x6f, 0x72, 0x79, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, + 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, + 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70, + 0x6f, 0x6e, 0x73, 0x65, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x22, 0x3b, 0x0a, 0x0f, 0x53, 0x65, + 0x74, 0x43, 0x50, 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, - 0x65, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x22, 0x3b, 0x0a, 0x0f, 0x53, 0x65, 0x74, 0x43, 0x50, - 0x55, 0x4c, 0x69, 0x6d, 0x69, 0x74, 0x52, 0x65, 0x73, 0x70, 0x12, 0x28, 0x0a, 0x04, 0x42, 0x61, - 0x73, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x63, 0x6f, 0x6d, 0x6d, 0x6f, - 0x6e, 0x2e, 0x42, 0x61, 0x73, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x52, 0x04, - 0x42, 0x61, 0x73, 0x65, 0x2a, 0x4e, 0x0a, 0x0a, 0x4c, 0x6f, 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, - 0x6f, 0x6e, 0x12, 0x0e, 0x0a, 0x0a, 0x4c, 0x6f, 0x61, 0x64, 0x56, 0x65, 0x72, 0x74, 0x65, 0x78, - 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x4c, 0x6f, 0x61, 0x64, 0x53, 0x63, 0x61, 0x74, 0x74, 0x65, - 0x72, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x4c, 0x6f, 0x61, 0x64, 0x45, 0x64, 0x67, 0x65, 0x10, - 0x02, 0x12, 0x11, 0x0a, 0x0d, 0x4c, 0x6f, 0x61, 0x64, 0x4f, 0x75, 0x74, 0x44, 0x65, 0x67, 0x72, - 0x65, 0x65, 0x10, 0x03, 0x2a, 0x32, 0x0a, 0x0d, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, - 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0f, 0x0a, 0x0b, 0x53, 0x65, 0x74, 0x4f, 0x75, 0x74, 0x45, - 0x64, 0x67, 0x65, 0x73, 0x10, 0x00, 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x65, 0x74, 0x4f, 0x75, 0x74, - 0x44, 0x65, 0x67, 0x72, 0x65, 0x65, 0x10, 0x01, 0x32, 0xb7, 0x07, 0x0a, 0x06, 0x57, 0x6f, 0x72, - 0x6b, 0x65, 0x72, 0x12, 0x3d, 0x0a, 0x0c, 0x53, 0x61, 0x79, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x50, - 0x65, 0x65, 0x72, 0x12, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x48, 0x65, 0x6c, - 0x6c, 0x6f, 0x50, 0x65, 0x65, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x15, 0x2e, 0x6d, 0x61, 0x73, 0x74, - 0x65, 0x72, 0x2e, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x50, 0x65, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, - 0x22, 0x00, 0x12, 0x38, 0x0a, 0x07, 0x53, 0x63, 0x61, 0x74, 0x74, 0x65, 0x72, 0x12, 0x12, 0x2e, - 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x61, 0x74, 0x74, 0x65, 0x72, 0x52, 0x65, - 0x71, 0x1a, 0x13, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x61, 0x74, 0x74, - 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x41, 0x0a, 0x0a, - 0x4c, 0x6f, 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x2e, 0x6d, 0x61, 0x73, - 0x74, 0x65, 0x72, 0x2e, 0x4c, 0x6f, 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, - 0x71, 0x1a, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x4c, 0x6f, 0x61, 0x64, 0x41, - 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, - 0x37, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x45, 0x64, 0x67, 0x65, 0x73, 0x12, 0x13, 0x2e, 0x6d, 0x61, - 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x45, 0x64, 0x67, 0x65, 0x73, 0x52, 0x65, 0x71, - 0x1a, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x45, 0x64, 0x67, - 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x09, 0x47, 0x65, 0x74, 0x56, - 0x65, 0x72, 0x74, 0x65, 0x78, 0x12, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, - 0x65, 0x74, 0x56, 0x65, 0x72, 0x74, 0x65, 0x78, 0x52, 0x65, 0x71, 0x1a, 0x15, 0x2e, 0x6d, 0x61, - 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x74, 0x65, 0x78, 0x52, 0x65, - 0x73, 0x70, 0x22, 0x00, 0x12, 0x40, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x47, 0x72, - 0x61, 0x70, 0x68, 0x12, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x44, 0x65, 0x6c, - 0x65, 0x74, 0x65, 0x47, 0x72, 0x61, 0x70, 0x68, 0x52, 0x65, 0x71, 0x1a, 0x17, 0x2e, 0x6d, 0x61, - 0x73, 0x74, 0x65, 0x72, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x47, 0x72, 0x61, 0x70, 0x68, - 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x38, 0x0a, 0x07, 0x53, 0x74, 0x65, 0x70, 0x45, 0x6e, - 0x64, 0x12, 0x12, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x74, 0x65, 0x70, 0x45, - 0x6e, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x13, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, - 0x74, 0x65, 0x70, 0x45, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, - 0x12, 0x40, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x54, 0x61, 0x73, 0x6b, 0x12, - 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, - 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x1a, 0x17, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, - 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x73, 0x70, - 0x22, 0x00, 0x12, 0x48, 0x0a, 0x09, 0x53, 0x61, 0x76, 0x65, 0x47, 0x72, 0x61, 0x70, 0x68, 0x12, - 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, 0x65, - 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x1a, 0x1c, 0x2e, 0x6d, + 0x65, 0x52, 0x04, 0x42, 0x61, 0x73, 0x65, 0x2a, 0x4e, 0x0a, 0x0a, 0x4c, 0x6f, 0x61, 0x64, 0x41, + 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0e, 0x0a, 0x0a, 0x4c, 0x6f, 0x61, 0x64, 0x56, 0x65, 0x72, + 0x74, 0x65, 0x78, 0x10, 0x00, 0x12, 0x0f, 0x0a, 0x0b, 0x4c, 0x6f, 0x61, 0x64, 0x53, 0x63, 0x61, + 0x74, 0x74, 0x65, 0x72, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x4c, 0x6f, 0x61, 0x64, 0x45, 0x64, + 0x67, 0x65, 0x10, 0x02, 0x12, 0x11, 0x0a, 0x0d, 0x4c, 0x6f, 0x61, 0x64, 0x4f, 0x75, 0x74, 0x44, + 0x65, 0x67, 0x72, 0x65, 0x65, 0x10, 0x03, 0x2a, 0x32, 0x0a, 0x0d, 0x53, 0x65, 0x74, 0x74, 0x69, + 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0f, 0x0a, 0x0b, 0x53, 0x65, 0x74, 0x4f, + 0x75, 0x74, 0x45, 0x64, 0x67, 0x65, 0x73, 0x10, 0x00, 0x12, 0x10, 0x0a, 0x0c, 0x53, 0x65, 0x74, + 0x4f, 0x75, 0x74, 0x44, 0x65, 0x67, 0x72, 0x65, 0x65, 0x10, 0x01, 0x32, 0xb7, 0x07, 0x0a, 0x06, + 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x3d, 0x0a, 0x0c, 0x53, 0x61, 0x79, 0x48, 0x65, 0x6c, + 0x6c, 0x6f, 0x50, 0x65, 0x65, 0x72, 0x12, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, + 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x50, 0x65, 0x65, 0x72, 0x52, 0x65, 0x71, 0x1a, 0x15, 0x2e, 0x6d, + 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x50, 0x65, 0x65, 0x72, 0x52, + 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x38, 0x0a, 0x07, 0x53, 0x63, 0x61, 0x74, 0x74, 0x65, 0x72, + 0x12, 0x12, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x63, 0x61, 0x74, 0x74, 0x65, + 0x72, 0x52, 0x65, 0x71, 0x1a, 0x13, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x63, + 0x61, 0x74, 0x74, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, + 0x41, 0x0a, 0x0a, 0x4c, 0x6f, 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x2e, + 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x4c, 0x6f, 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, 0x6f, + 0x6e, 0x52, 0x65, 0x71, 0x1a, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x4c, 0x6f, + 0x61, 0x64, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, + 0x30, 0x01, 0x12, 0x37, 0x0a, 0x08, 0x47, 0x65, 0x74, 0x45, 0x64, 0x67, 0x65, 0x73, 0x12, 0x13, + 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x45, 0x64, 0x67, 0x65, 0x73, + 0x52, 0x65, 0x71, 0x1a, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, + 0x45, 0x64, 0x67, 0x65, 0x73, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x3a, 0x0a, 0x09, 0x47, + 0x65, 0x74, 0x56, 0x65, 0x72, 0x74, 0x65, 0x78, 0x12, 0x14, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x74, 0x65, 0x78, 0x52, 0x65, 0x71, 0x1a, 0x15, + 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x65, 0x74, 0x56, 0x65, 0x72, 0x74, 0x65, + 0x78, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x40, 0x0a, 0x0b, 0x44, 0x65, 0x6c, 0x65, 0x74, + 0x65, 0x47, 0x72, 0x61, 0x70, 0x68, 0x12, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, + 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x47, 0x72, 0x61, 0x70, 0x68, 0x52, 0x65, 0x71, 0x1a, 0x17, + 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x47, 0x72, + 0x61, 0x70, 0x68, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x38, 0x0a, 0x07, 0x53, 0x74, 0x65, + 0x70, 0x45, 0x6e, 0x64, 0x12, 0x12, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x74, + 0x65, 0x70, 0x45, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x1a, 0x13, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x53, 0x74, 0x65, 0x70, 0x45, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, + 0x01, 0x30, 0x01, 0x12, 0x40, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x54, 0x61, + 0x73, 0x6b, 0x12, 0x16, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x43, 0x6f, 0x6e, 0x74, + 0x72, 0x6f, 0x6c, 0x54, 0x61, 0x73, 0x6b, 0x52, 0x65, 0x71, 0x1a, 0x17, 0x2e, 0x6d, 0x61, 0x73, + 0x74, 0x65, 0x72, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x54, 0x61, 0x73, 0x6b, 0x52, + 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x48, 0x0a, 0x09, 0x53, 0x61, 0x76, 0x65, 0x47, 0x72, 0x61, + 0x70, 0x68, 0x12, 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, + 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x1a, + 0x1c, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, 0x65, + 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, + 0x48, 0x0a, 0x09, 0x57, 0x72, 0x69, 0x74, 0x65, 0x44, 0x69, 0x73, 0x6b, 0x12, 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, - 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x48, 0x0a, 0x09, - 0x57, 0x72, 0x69, 0x74, 0x65, 0x44, 0x69, 0x73, 0x6b, 0x12, 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, + 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x1a, 0x1c, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, - 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, 0x1a, 0x1c, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, + 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x48, 0x0a, 0x09, 0x52, 0x65, 0x61, + 0x64, 0x47, 0x72, 0x61, 0x70, 0x68, 0x12, 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, - 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x48, 0x0a, 0x09, 0x52, 0x65, 0x61, 0x64, 0x47, 0x72, - 0x61, 0x70, 0x68, 0x12, 0x1b, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, - 0x70, 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x71, - 0x1a, 0x1c, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, 0x70, 0x68, 0x50, - 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, - 0x12, 0x4c, 0x0a, 0x11, 0x47, 0x65, 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, - 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x57, - 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x71, - 0x1a, 0x1a, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, - 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x12, 0x4a, - 0x0a, 0x0d, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, - 0x18, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, - 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x1a, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, - 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, - 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x46, 0x0a, 0x0d, 0x52, 0x75, - 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x18, 0x2e, 0x6d, 0x61, - 0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, - 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x1a, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, - 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, - 0x22, 0x00, 0x42, 0x03, 0x5a, 0x01, 0x2f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x52, 0x65, 0x71, 0x1a, 0x1c, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x47, 0x72, 0x61, + 0x70, 0x68, 0x50, 0x65, 0x72, 0x73, 0x69, 0x73, 0x74, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x65, 0x73, + 0x70, 0x22, 0x00, 0x12, 0x4c, 0x0a, 0x11, 0x47, 0x65, 0x74, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, + 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, + 0x52, 0x65, 0x71, 0x1a, 0x1a, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x57, 0x6f, 0x72, + 0x6b, 0x65, 0x72, 0x53, 0x74, 0x61, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x22, + 0x00, 0x12, 0x4a, 0x0a, 0x0d, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, + 0x6f, 0x6e, 0x12, 0x18, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, + 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x1a, 0x19, 0x2e, 0x6d, + 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x53, 0x65, 0x74, 0x74, 0x69, 0x6e, 0x67, 0x41, 0x63, 0x74, + 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x46, 0x0a, + 0x0d, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x18, + 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, + 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x1a, 0x19, 0x2e, 0x6d, 0x61, 0x73, 0x74, 0x65, + 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, + 0x65, 0x73, 0x70, 0x22, 0x00, 0x42, 0x03, 0x5a, 0x01, 0x2f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, + 0x6f, 0x33, } var ( diff --git a/vermeer/apps/protos/worker.proto b/vermeer/apps/protos/worker.proto index 0468967bd..eea5fbee4 100644 --- a/vermeer/apps/protos/worker.proto +++ b/vermeer/apps/protos/worker.proto @@ -197,6 +197,14 @@ message WorkerStatInfoResp{ common.BaseResponse Base = 1; double MemMachineUsedPercent = 2; double MemProcessUsedPercent = 3; + // Process resource snapshots. CPU is consumed CPU seconds / elapsed seconds, + // measured in cores, and RSS is absolute resident bytes, not host/Pod ratios. + uint32 metric_schema_version = 4; + double process_cpu_cores = 5; + uint64 process_rss_bytes = 6; + double sample_window_seconds = 7; + int64 metrics_observed_at = 8; + string metric_error = 9; } enum SettingAction @@ -273,4 +281,4 @@ message SetMemoryLimitResp { message SetCPULimitResp { common.BaseResponse Base = 1; -} \ No newline at end of file +} diff --git a/vermeer/apps/worker/grpc_handlers.go b/vermeer/apps/worker/grpc_handlers.go index 9bb3bb5e4..c4a8366ee 100644 --- a/vermeer/apps/worker/grpc_handlers.go +++ b/vermeer/apps/worker/grpc_handlers.go @@ -249,7 +249,7 @@ func (ph *PeerHandler) GetWorkerStatInfo(ctx context.Context, req *pb.WorkerStat _ = ctx _ = req - resp := new(pb.WorkerStatInfoResp) + resp := &pb.WorkerStatInfoResp{Base: &pb.BaseResponse{}, MetricSchemaVersion: 1} mp, err := common.MachineMemUsedPercent() if err != nil { resp.Base.ErrorCode = -1 @@ -264,8 +264,17 @@ func (ph *PeerHandler) GetWorkerStatInfo(ctx context.Context, req *pb.WorkerStat } resp.MemProcessUsedPercent = pp - //logrus.Infof("GetWorkerStatInfo mem:%v", common.PrintMemUsage()) - //logrus.Infof("GetWorkerStatInfo res:%v", resp) + metrics, metricErr := common.ReadProcessMetrics() + resp.MetricsObservedAt = metrics.ObservedAt + if metricErr != nil { + // Resource collection is independent of Worker readiness and legacy + // memory statistics; a failed sample must not become a service error. + resp.MetricError = metricErr.Error() + } else { + resp.ProcessCpuCores = metrics.CPUCores + resp.ProcessRssBytes = metrics.RSSBytes + resp.SampleWindowSeconds = metrics.SampleWindowSeconds + } return resp, nil diff --git a/vermeer/apps/worker/process_metrics_test.go b/vermeer/apps/worker/process_metrics_test.go new file mode 100644 index 000000000..6b1e673cf --- /dev/null +++ b/vermeer/apps/worker/process_metrics_test.go @@ -0,0 +1,62 @@ +/* +Licensed to the Apache Software Foundation (ASF) under one or more +contributor license agreements. See the NOTICE file distributed with this +work for additional information regarding copyright ownership. The ASF +licenses this file to You under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +License for the specific language governing permissions and limitations +under the License. +*/ + +package worker + +import ( + "context" + "net" + "testing" + "time" + pb "vermeer/apps/protos" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" +) + +func TestProcessMetricsTravelThroughWorkerRPC(t *testing.T) { + listener := bufconn.Listen(1024 * 1024) + server := grpc.NewServer() + pb.RegisterWorkerServer(server, &PeerHandler{}) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + connection, err := grpc.DialContext(ctx, "passthrough:///worker-test", + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), + grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = connection.Close() }) + response, err := pb.NewWorkerClient(connection).GetWorkerStatInfo(ctx, &pb.WorkerStatInfoReq{}) + if err != nil { + t.Fatal(err) + } + if response.GetBase() == nil || response.GetMetricSchemaVersion() != 1 || response.GetMetricsObservedAt() <= 0 { + t.Fatalf("missing resource protocol metadata: %v", response) + } + if response.GetMetricError() == "" { + if response.GetProcessRssBytes() == 0 || response.GetSampleWindowSeconds() <= 0 { + t.Fatalf("successful resource snapshot needs RSS and a real CPU window: %v", response) + } + } else if response.GetMetricError() != "metrics_warming_up" && + response.GetMetricError() != "process_metrics_unavailable" { + t.Fatalf("unexpected collection failure: %s", response.GetMetricError()) + } +} From 8ebb147f59c26b41c10abfcb3df1c06a80d686ac Mon Sep 17 00:00:00 2001 From: orion Date: Thu, 29 Oct 2026 12:29:11 +0000 Subject: [PATCH 2/2] fix(vermeer): stop telemetry queue after cancellation --- vermeer/apps/master/services/http_cluster.go | 54 ++++++++- .../apps/master/services/http_cluster_test.go | 111 ++++++++++++++++++ 2 files changed, 159 insertions(+), 6 deletions(-) diff --git a/vermeer/apps/master/services/http_cluster.go b/vermeer/apps/master/services/http_cluster.go index f930e5f8a..0e4c76589 100644 --- a/vermeer/apps/master/services/http_cluster.go +++ b/vermeer/apps/master/services/http_cluster.go @@ -77,37 +77,64 @@ func collectClusterSnapshot(parent context.Context) ClusterSnapshot { now := time.Now().UnixMilli() snapshots := WorkerMgr.HealthSnapshots() info := InfoMaster.snapshot() + return buildClusterSnapshot(parent, info, snapshots, now) +} + +func buildClusterSnapshot(parent context.Context, info MasterInfoSnapshot, + snapshots []workers.WorkerHealthSnapshot, now int64) ClusterSnapshot { result := ClusterSnapshot{ SchemaVersion: 1, ObservedAt: now, Status: "UP", RegisteredWorkers: len(snapshots), Master: clusterMasterNode(info, now), Workers: make([]ClusterNode, len(snapshots)), } + for index, snapshot := range snapshots { + result.Workers[index] = clusterWorkerMetadata(snapshot, now) + } // One total deadline and bounded fanout prevent telemetry from delaying health // indefinitely. Worker metadata is already copied, so no lock spans an RPC. ctx, cancel := context.WithTimeout(parent, 800*time.Millisecond) defer cancel() var wg sync.WaitGroup - jobs := make(chan int) count := len(snapshots) if count > 8 { count = 8 } + jobs := make(chan int, count) for i := 0; i < count; i++ { wg.Add(1) go func() { defer wg.Done() - for index := range jobs { - result.Workers[index] = clusterWorkerNode(ctx, snapshots[index], now) + for { + select { + case <-ctx.Done(): + return + case index, ok := <-jobs: + if !ok || ctx.Err() != nil { + return + } + result.Workers[index] = collectWorkerMetrics(ctx, snapshots[index], result.Workers[index], now) + } } }() } +enqueue: for index := range snapshots { - jobs <- index + if ctx.Err() != nil { + break + } + select { + case <-ctx.Done(): + break enqueue + case jobs <- index: + } } close(jobs) wg.Wait() - for _, worker := range result.Workers { + for index, worker := range result.Workers { + if ctx.Err() != nil && worker.MetricStatus.Reason == "metrics_not_collected" { + result.Workers[index].MetricStatus.Reason = "metrics_timeout" + } if worker.Ready { result.ReadyWorkers++ } @@ -151,6 +178,10 @@ func clusterMasterNode(info MasterInfoSnapshot, now int64) ClusterNode { } func clusterWorkerNode(ctx context.Context, snapshot workers.WorkerHealthSnapshot, now int64) ClusterNode { + return collectWorkerMetrics(ctx, snapshot, clusterWorkerMetadata(snapshot, now), now) +} + +func clusterWorkerMetadata(snapshot workers.WorkerHealthSnapshot, now int64) ClusterNode { checks := snapshot.Checks node := ClusterNode{ID: snapshot.Name, Name: snapshot.Name, Role: "WORKER", Group: snapshot.Group, GrpcPeer: snapshot.GrpcPeer, Version: snapshot.Version, @@ -165,7 +196,18 @@ func clusterWorkerNode(ctx context.Context, snapshot workers.WorkerHealthSnapsho if !node.Ready { node.Status = "DEGRADED" } - if !checks.ConnectionReady || snapshot.Session == nil { + if checks.ConnectionReady && snapshot.Session != nil { + node.MetricStatus.Reason = "metrics_not_collected" + } + return node +} + +func collectWorkerMetrics(ctx context.Context, snapshot workers.WorkerHealthSnapshot, node ClusterNode, now int64) ClusterNode { + if !snapshot.Checks.ConnectionReady || snapshot.Session == nil { + return node + } + if ctx.Err() != nil { + node.MetricStatus = ClusterMetricStatus{"UNAVAILABLE", now, "metrics_timeout"} return node } response, err := snapshot.Session.GetWorkerStatInfo(ctx, &pb.WorkerStatInfoReq{}) diff --git a/vermeer/apps/master/services/http_cluster_test.go b/vermeer/apps/master/services/http_cluster_test.go index 0dfde6406..a29339218 100644 --- a/vermeer/apps/master/services/http_cluster_test.go +++ b/vermeer/apps/master/services/http_cluster_test.go @@ -21,12 +21,14 @@ import ( "context" "encoding/json" "errors" + "fmt" "github.com/gin-gonic/gin" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "math" "net/http/httptest" + "sync/atomic" "testing" "time" "vermeer/apps/auth" @@ -42,6 +44,115 @@ type clusterStatsClient struct { err error } +type queuedClusterStatsClient struct { + pb.WorkerClient + block bool + started chan struct{} + calls atomic.Int32 + canceledCalls atomic.Int32 +} + +func (client *queuedClusterStatsClient) GetWorkerStatInfo(ctx context.Context, _ *pb.WorkerStatInfoReq, + _ ...grpc.CallOption) (*pb.WorkerStatInfoResp, error) { + client.calls.Add(1) + if ctx.Err() != nil { + client.canceledCalls.Add(1) + } + client.started <- struct{}{} + if client.block { + <-ctx.Done() + return nil, status.FromContextError(ctx.Err()).Err() + } + return &pb.WorkerStatInfoResp{MetricSchemaVersion: 1, ProcessCpuCores: 0.5, + ProcessRssBytes: 2048, SampleWindowSeconds: 1, MetricsObservedAt: 1234}, nil +} + +func TestClusterMetricQueuePreservesAllWorkerHealth(t *testing.T) { + const total = 25 + for _, mode := range []string{"success", "already_canceled", "cancel_in_flight", "deadline"} { + t.Run(mode, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if mode == "already_canceled" { + cancel() + } + client := &queuedClusterStatsClient{ + block: mode == "cancel_in_flight" || mode == "deadline", + started: make(chan struct{}, total), + } + snapshots := make([]workers.WorkerHealthSnapshot, total) + for index := range snapshots { + snapshot := readyWorkerSnapshot(client) + snapshot.Name = fmt.Sprintf("worker-%02d", index) + snapshot.GrpcPeer = fmt.Sprintf("worker-%02d:6789", index) + snapshot.Version = "test-version" + snapshot.LaunchTime = time.Unix(5, 0) + snapshots[index] = snapshot + } + // A readiness failure must remain distinct from a metric timeout, even + // for a node queued beyond the first eight in-flight RPCs. + snapshots[total-1].Checks.ComputeStreamReady = false + done := make(chan ClusterSnapshot, 1) + started := time.Now() + go func() { + done <- buildClusterSnapshot(ctx, MasterInfoSnapshot{LaunchTime: time.Unix(1, 0)}, snapshots, 1234) + }() + if mode == "cancel_in_flight" { + for i := 0; i < 8; i++ { + select { + case <-client.started: + case <-time.After(2 * time.Second): + t.Fatal("metrics pool did not start eight RPCs") + } + } + cancel() + } + var result ClusterSnapshot + select { + case result = <-done: + case <-time.After(3 * time.Second): + t.Fatal("metrics queue did not stop after cancellation/deadline") + } + if time.Since(started) > 2*time.Second { + t.Fatal("metrics queue added work beyond its shared deadline") + } + wantCalls := int32(8) + if mode == "success" { + wantCalls = total + } + if mode == "already_canceled" { + wantCalls = 0 + } + if client.calls.Load() != wantCalls || client.canceledCalls.Load() != 0 { + t.Fatalf("calls=%d, want=%d; calls with canceled context=%d", + client.calls.Load(), wantCalls, client.canceledCalls.Load()) + } + if result.RegisteredWorkers != total || result.ReadyWorkers != total-1 || len(result.Workers) != total || + result.Status != "DEGRADED" || result.Reason != "workers_not_ready" { + t.Fatalf("metric collection changed aggregate health: %+v", result) + } + for index, node := range result.Workers { + snapshot := snapshots[index] + if node.ID != snapshot.Name || node.Name != snapshot.Name || node.Role != "WORKER" || + node.Group != snapshot.Group || node.GrpcPeer != snapshot.GrpcPeer || node.Version != snapshot.Version || + !node.LaunchTime.Equal(snapshot.LaunchTime) || node.Ready != snapshot.Checks.Ready() || + node.Reason != snapshot.Checks.Reason() || len(node.Checks) != 5 || + node.Checks["compute_stream_ready"] != snapshot.Checks.ComputeStreamReady || node.Metrics == nil { + t.Fatalf("worker %d lost health metadata: %+v", index, node) + } + if mode == "success" { + if node.MetricStatus.Availability != "AVAILABLE" || node.Metrics["rss_bytes"] != uint64(2048) { + t.Fatalf("worker %d missing collected metrics: %+v", index, node) + } + } else if node.MetricStatus.Availability != "UNAVAILABLE" || node.MetricStatus.Reason != "metrics_timeout" || + len(node.Metrics) != 0 { + t.Fatalf("worker %d missing timeout status: %+v", index, node) + } + } + }) + } +} + func (client clusterStatsClient) GetWorkerStatInfo(context.Context, *pb.WorkerStatInfoReq, ...grpc.CallOption) (*pb.WorkerStatInfoResp, error) { return client.response, client.err }