hanahmily commented on code in PR #1164:
URL:
https://github.com/apache/skywalking-banyandb/pull/1164#discussion_r3372702540
##########
banyand/backup/lifecycle/service.go:
##########
@@ -245,10 +313,77 @@ func (l *lifecycleService) Name() string {
return "lifecycle"
}
+// buildLocalNodeMD builds the schema metadata registered on the native metrics
+// pub client for the co-located data node. The ROLE_DATA role is REQUIRED: the
+// pub client's role gate (allowedRoles=[ROLE_DATA]) silently drops a node
whose
+// roles do not intersect, which would make native metrics a silent no-op.
+func buildLocalNodeMD(grpcAddr string) schema.Metadata {
+ return schema.Metadata{
+ TypeMeta: schema.TypeMeta{Kind: schema.KindNode},
+ Spec: &databasev1.Node{
+ Metadata: &commonv1.Metadata{Name:
metricsLocalNodeName},
+ GrpcAddress: grpcAddr,
+ Roles:
[]databasev1.Role{databasev1.Role_ROLE_DATA},
+ },
+ }
+}
+
+// buildHTTPRouter assembles the lifecycle's HTTP routes: the gRPC-gateway API
+// under /api and, when the metrics registry exposes a Prometheus handler, the
+// /metrics endpoint on the same (existing) HTTP server.
+func (l *lifecycleService) buildHTTPRouter(apiHandler http.Handler) *chi.Mux {
+ mux := chi.NewRouter()
+ mux.Mount("/api", http.StripPrefix("/api", apiHandler))
+ if reg, ok := l.omr.(observability.PrometheusHandlerProvider); ok {
+ mux.Handle("/metrics", reg.PrometheusHandler())
+ }
+ return mux
+}
+
+// startMetricsNodeKeeper registers the co-located data node on the native
+// metrics pub client and keeps the connection active. The pub connManager
runs a
+// synchronous health check at registration and does NOT auto-retry an evicted
+// node (NewWithoutMetadata leaves healthCheckInterval==0), so the keeper
+// re-registers whenever the node is not locatable — covering both an initial
+// dial that failed (data node not ready yet) and a later failover eviction.
+func (l *lifecycleService) startMetricsNodeKeeper() {
+ l.localNodeMD = buildLocalNodeMD(l.gRPCAddr)
+ l.metricsKeeperStop = make(chan struct{})
+ // Initial registration dials the local data node so native flushes
route there.
+ l.metricsClient.OnAddOrUpdate(l.localNodeMD)
+ run.Go(context.Background(), "backup.lifecycle.metrics-node-keeper",
l.l, func(_ context.Context) {
+ ticker := time.NewTicker(metricsNodeKeeperInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-l.metricsKeeperStop:
+ return
+ case <-ticker.C:
+ if _, locateErr :=
l.metricsNodeRegistry.Locate(native.ObservabilityGroupName, "", 0, 0);
locateErr != nil {
+ // connMgr.OnAddOrUpdate short-circuits
an evictable node, so drop
+ // it first to force a fresh dial.
queue.Client does not expose
+ // OnDelete, hence the type assertion.
+ if deleter, ok :=
l.metricsClient.(interface{ OnDelete(schema.Metadata) }); ok {
+ deleter.OnDelete(l.localNodeMD)
+ }
+
l.metricsClient.OnAddOrUpdate(l.localNodeMD)
+ }
+ }
+ }
+ })
+}
Review Comment:
Switched the keeper to probe `metricsClient.HealthyNodes()` (the pub's
active-connection state) instead of `Locate()` in f76690ea, and dropped the
now-unused `metricsNodeRegistry` field — thanks for the nudge on the health
signal.
On the seeding: `connMgr.OnAddOrUpdate` synchronously calls
`handler.OnActive` (connmanager.go:180), and `pub.OnActive` fans out
`OnAddOrUpdate` to registered handlers, so `clusterNodeService.OnAddOrUpdate` →
`sel.AddNode` does seed the selector once the node dials active (the same
selector path the migration routing exercises in the distributed integration
test). The keeper re-registers via OnDelete+OnAddOrUpdate whenever
HealthyNodes() is empty, covering the not-yet-reachable case.
##########
banyand/observability/services/service.go:
##########
@@ -78,6 +80,17 @@ func NewMetricService(metadata metadata.Repo, pipeline
queue.Client, nodeType st
}
}
+// NewMetricServiceWithoutListener returns a metric service that does not start
+// its own HTTP listener. The embedding service is responsible for serving the
+// metrics, e.g. by mounting PrometheusHandler on its own router. This lets the
+// lifecycle command expose /metrics on its existing HTTP port instead of
opening
+// a separate observability listener.
+func NewMetricServiceWithoutListener(metadata metadata.Repo, pipeline
queue.Client, nodeType string, nodeSelector native.NodeSelector)
observability.MetricsRegistry {
+ svc, _ := NewMetricService(metadata, pipeline, nodeType,
nodeSelector).(*metricService)
+ svc.listenerDisabled = true
+ return svc
Review Comment:
Done in f76690ea — switched to a direct type assertion (`svc :=
NewMetricService(...).(*metricService)`) so a future change to
NewMetricService's return type fails with a clear interface-conversion panic
instead of a nil dereference.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
