* Copyright (c) 2026 Huawei Technologies Co., Ltd.
* openFuyao is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
* http://license.coscl.org.cn/MulanPSL2
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
*/
package analyzer
import (
"errors"
"io"
"math"
"time"
"many-core-orchestrator/pkg/api/analyzerv1"
"google.golang.org/grpc"
olog "gopkg.openfuyao.cn/common-modules/ologger/log"
)
type AnalyzerService struct {
analyzerv1.UnimplementedAnalyzerServer
store *NodeStateStore
logger *olog.Logger
}
func NewAnalyzerService(store *NodeStateStore, logger *olog.Logger) *AnalyzerService {
return &AnalyzerService{store: store, logger: logger}
}
func (s *AnalyzerService) CollectMetrics(stream grpc.ClientStreamingServer[analyzerv1.MetricsSnapshot, analyzerv1.CollectMetricsResponse]) error {
grpcRequestsTotal.WithLabelValues("CollectMetrics", "started").Inc()
var accepted uint64
for {
snapshot, err := stream.Recv()
if errors.Is(err, io.EOF) {
grpcRequestsTotal.WithLabelValues("CollectMetrics", "ok").Inc()
return stream.SendAndClose(&analyzerv1.CollectMetricsResponse{Accepted: accepted})
}
if err != nil {
grpcRequestsTotal.WithLabelValues("CollectMetrics", "error").Inc()
return err
}
if snapshot.GetNode() == "" || snapshot.GetMetrics() == nil {
continue
}
start := time.Now()
metrics := sanitizeMetrics(snapshot.GetNode(), metricsFromProto(snapshot.GetMetrics()))
report := s.store.AddWithMetadata(snapshot.GetNode(), parseTimestamp(snapshot.GetTimestamp()), metrics, snapshot.GetStatus(), metadataFromProto(snapshot))
nodeInterferenceLevel.WithLabelValues(report.NodeName).Set(report.InterferenceLevel)
collectorSamplesTotal.WithLabelValues(report.NodeName).Inc()
reportGenerationDuration.Observe(time.Since(start).Seconds())
accepted++
}
}
func metadataFromProto(snapshot *analyzerv1.MetricsSnapshot) ReportMetadata {
availability := make([]MetricAvailability, 0, len(snapshot.GetAvailability()))
for _, item := range snapshot.GetAvailability() {
availability = append(availability, MetricAvailability{
Collector: item.GetCollector(),
Metric: item.GetMetric(),
Available: item.GetAvailable(),
Status: item.GetStatus(),
Reason: item.GetReason(),
})
}
return ReportMetadata{
Sequence: snapshot.GetSequence(),
Availability: availability,
}
}
func sanitizeMetrics(node string, metrics NodeMetrics) NodeMetrics {
return NodeMetrics{
LLCMissRate: sanitizeMetricValue(node, "llcMissRate", metrics.LLCMissRate),
LLCOccupancy: sanitizeMetricValue(node, "llcOccupancy", metrics.LLCOccupancy),
BioLatencyP95Ms: sanitizeMetricValue(node, "bioLatencyP95Ms", metrics.BioLatencyP95Ms),
BioLatencyP99Ms: sanitizeMetricValue(node, "bioLatencyP99Ms", metrics.BioLatencyP99Ms),
PSIAvailable: metrics.PSIAvailable,
PSIIOAvg10: sanitizeMetricValue(node, "psiIoSome", metrics.PSIIOAvg10),
PSIIOFullAvg10: sanitizeMetricValue(node, "psiIoFull", metrics.PSIIOFullAvg10),
PSIMemAvg10: sanitizeMetricValue(node, "psiMemorySome", metrics.PSIMemAvg10),
PSIMemFullAvg10: sanitizeMetricValue(node, "psiMemoryFull", metrics.PSIMemFullAvg10),
MemAvailRatio: sanitizeMetricValue(node, "memAvailableRatio", metrics.MemAvailRatio),
PageScanRate: sanitizeMetricValue(node, "pageScanRate", metrics.PageScanRate),
}
}
func sanitizeMetricValue(node, metric string, value float64) float64 {
switch {
case math.IsNaN(value):
invalidMetricValuesTotal.WithLabelValues(node, metric, "nan").Inc()
return 0
case math.IsInf(value, 0):
invalidMetricValuesTotal.WithLabelValues(node, metric, "inf").Inc()
return 0
default:
return value
}
}
func parseTimestamp(value string) time.Time {
if value == "" {
return time.Now().UTC()
}
if parsed, err := time.Parse(time.RFC3339Nano, value); err == nil {
return parsed.UTC()
}
return time.Now().UTC()
}
func metricsFromProto(metrics *analyzerv1.NodeMetrics) NodeMetrics {
return NodeMetrics{
LLCMissRate: metrics.GetLlcMissRate(),
LLCOccupancy: metrics.GetLlcOccupancy(),
BioLatencyP95Ms: metrics.GetBioLatencyP95Ms(),
BioLatencyP99Ms: metrics.GetBioLatencyP99Ms(),
PSIAvailable: metrics.GetPsiAvailable(),
PSIIOAvg10: metrics.GetPsiIoSome(),
PSIIOFullAvg10: metrics.GetPsiIoFull(),
PSIMemAvg10: metrics.GetPsiMemorySome(),
PSIMemFullAvg10: metrics.GetPsiMemoryFull(),
MemAvailRatio: metrics.GetMemAvailableRatio(),
PageScanRate: metrics.GetPageScanRate(),
}
}