feat(vermeer): expose independent cluster readiness and process telemetry

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.

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
 }
 

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
+}

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()
+}