* 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 (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"sync/atomic"
"time"
"many-core-orchestrator/api/v1alpha1"
olog "gopkg.openfuyao.cn/common-modules/ologger/log"
"k8s.io/apimachinery/pkg/runtime/schema"
"sigs.k8s.io/yaml"
)
type RuleEngine interface {
Evaluate(metrics NodeMetrics) []Detection
}
type InterferenceCalculator interface {
Compute(metrics NodeMetrics) float64
}
type DetectionRules struct {
Weights RuleWeights `json:"weights"`
Saturation RuleSaturation `json:"saturation"`
Thresholds RuleThresholds `json:"thresholds"`
Penalties RulePenalties `json:"penalties"`
DegradedMemory RuleDegradedMemory `json:"degradedMemory"`
}
type RuleWeights struct {
Cache float64 `json:"cache"`
Memory float64 `json:"memory"`
IO float64 `json:"io"`
}
type RuleSaturation struct {
CacheProduct float64 `json:"cacheProduct"`
PSIMemory float64 `json:"psiMemory"`
BioLatencyP99Ms float64 `json:"bioLatencyP99Ms"`
PSIIO float64 `json:"psiIo"`
}
type RuleThresholds struct {
CacheMissRate float64 `json:"cacheMissRate"`
LLCOccupancy float64 `json:"llcOccupancy"`
BioLatencyP99Ms float64 `json:"bioLatencyP99Ms"`
PSIIO float64 `json:"psiIo"`
PSIMemorySome float64 `json:"psiMemorySome"`
PSIMemoryFull float64 `json:"psiMemoryFull"`
IOPenaltyP99Ms float64 `json:"ioPenaltyP99Ms"`
IOPenaltyPSIIO float64 `json:"ioPenaltyPsiIo"`
MemIOPenaltyMemory float64 `json:"memIoPenaltyMemory"`
MemIOPenaltyPSIIO float64 `json:"memIoPenaltyPsiIo"`
}
type RulePenalties struct {
IOMultiplier float64 `json:"ioMultiplier"`
MemIOMultiplier float64 `json:"memIoMultiplier"`
}
type RuleDegradedMemory struct {
MemAvailableRatioThreshold float64 `json:"memAvailableRatioThreshold"`
PageScanRateThreshold float64 `json:"pageScanRateThreshold"`
}
func DefaultDetectionRules() DetectionRules {
return DetectionRules{
Weights: RuleWeights{
Cache: 0.30,
Memory: 0.30,
IO: 0.40,
},
Saturation: RuleSaturation{
CacheProduct: 0.175,
PSIMemory: 50,
BioLatencyP99Ms: 100,
PSIIO: 20,
},
Thresholds: RuleThresholds{
CacheMissRate: 0.20,
LLCOccupancy: 0.70,
BioLatencyP99Ms: 50,
PSIIO: 20,
PSIMemorySome: 10,
PSIMemoryFull: 0,
IOPenaltyP99Ms: 50,
IOPenaltyPSIIO: 20,
MemIOPenaltyMemory: 50,
MemIOPenaltyPSIIO: 20,
},
Penalties: RulePenalties{
IOMultiplier: 1.2,
MemIOMultiplier: 1.1,
},
DegradedMemory: RuleDegradedMemory{
MemAvailableRatioThreshold: 0.30,
PageScanRateThreshold: 10000,
},
}
}
type RuleConfigStore struct {
logger *olog.Logger
resource schema.GroupVersionResource
namespace string
name string
current atomic.Value
lastHash atomic.Value
lastLoadedTime atomic.Value
}
func NewRuleConfigStore(cfg Config, logger *olog.Logger) *RuleConfigStore {
store := &RuleConfigStore{
logger: logger,
resource: schema.GroupVersionResource{
Group: cfg.DetectionRulesGroup,
Version: cfg.DetectionRulesVersion,
Resource: cfg.DetectionRulesResource,
},
namespace: cfg.DetectionRulesNamespace,
name: cfg.DetectionRulesName,
}
store.current.Store(DefaultDetectionRules())
return store
}
func (s *RuleConfigStore) Evaluate(metrics NodeMetrics) []Detection {
return s.rules().Evaluate(metrics)
}
func (s *RuleConfigStore) Compute(metrics NodeMetrics) float64 {
return s.rules().Compute(metrics)
}
func (s *RuleConfigStore) rules() DetectionRules {
rules, ok := s.current.Load().(DetectionRules)
if !ok {
return DefaultDetectionRules()
}
return rules
}
func (s *RuleConfigStore) Metadata() RuleConfigMetadata {
hash, _ := s.lastHash.Load().(string)
loadedAt, _ := s.lastLoadedTime.Load().(time.Time)
return RuleConfigMetadata{Hash: hash, LastLoadedTime: loadedAt}
}
type RuleConfigMetadata struct {
Hash string
LastLoadedTime time.Time
}
func (s *RuleConfigStore) ApplyDetectionRuleConfig(event string, ruleConfig *v1alpha1.DetectionRuleConfig) error {
if !s.isTargetRuleConfig(ruleConfig) {
return nil
}
rules, hash, err := s.rulesFromObject(ruleConfig)
if err != nil {
ruleReloadTotal.WithLabelValues("error").Inc()
s.logger.Error("reload detection rules CR failed; keeping previous valid config",
"error", err,
"event", event,
"namespace", s.namespace,
"name", s.name,
"resource", s.resource.Resource,
)
return err
}
s.current.Store(rules)
loadedAt := time.Now().UTC()
s.lastLoadedTime.Store(loadedAt)
ruleReloadTotal.WithLabelValues("success").Inc()
ruleLastReloadTimestamp.Set(float64(loadedAt.Unix()))
lastHash, _ := s.lastHash.Load().(string)
if hash != lastHash {
if lastHash != "" {
ruleConfigInfo.DeleteLabelValues(lastHash, s.ruleConfigRef())
}
ruleConfigInfo.WithLabelValues(hash, s.ruleConfigRef()).Set(1)
s.lastHash.Store(hash)
if lastHash == "" {
s.logger.Info("loaded detection rules CR",
"event", event,
"namespace", s.namespace,
"name", s.name,
"resource", s.resource.Resource,
"hash", hash,
"loaded_at", loadedAt.Format(time.RFC3339),
)
} else {
s.logger.Info("reloaded detection rules CR",
"event", event,
"namespace", s.namespace,
"name", s.name,
"resource", s.resource.Resource,
"hash", hash,
"previous_hash", lastHash,
"loaded_at", loadedAt.Format(time.RFC3339),
)
}
}
return nil
}
func (s *RuleConfigStore) KeepLastValidAfterDelete() {
s.logger.Warn("detection rules CR deleted; keeping previous valid config",
"namespace", s.namespace,
"name", s.name,
"resource", s.resource.Resource,
)
}
func (s *RuleConfigStore) rulesFromObject(ruleConfig *v1alpha1.DetectionRuleConfig) (DetectionRules, string, error) {
if !s.isTargetRuleConfig(ruleConfig) {
return DetectionRules{}, "", fmt.Errorf("ignore non-target rule config %s/%s", ruleConfig.GetNamespace(), ruleConfig.GetName())
}
data, err := json.Marshal(ruleConfig.Spec)
if err != nil {
return DetectionRules{}, "", err
}
rules, err := LoadDetectionRulesSpec(data)
if err != nil {
return DetectionRules{}, "", err
}
return rules, hashBytes(data), nil
}
func (s *RuleConfigStore) isTargetRuleConfig(obj *v1alpha1.DetectionRuleConfig) bool {
return obj.GetNamespace() == s.namespace && obj.GetName() == s.name
}
func LoadDetectionRulesSpec(data []byte) (DetectionRules, error) {
rules := DefaultDetectionRules()
if err := json.Unmarshal(data, &rules); err != nil {
return DetectionRules{}, err
}
rules.applyDefaults()
return rules, nil
}
func (s *RuleConfigStore) ruleConfigRef() string {
return fmt.Sprintf("%s/%s/%s.%s/%s", s.namespace, s.name, s.resource.Resource, s.resource.Version, s.resource.Group)
}
func LoadDetectionRules(path string) (DetectionRules, string, error) {
data, err := os.ReadFile(path)
if err != nil {
return DetectionRules{}, "", err
}
rules := DefaultDetectionRules()
if err := yaml.Unmarshal(data, &rules); err != nil {
return DetectionRules{}, "", err
}
rules.applyDefaults()
return rules, hashBytes(data), nil
}
func hashBytes(data []byte) string {
sum := sha256.Sum256(data)
return hex.EncodeToString(sum[:])
}
func (r *DetectionRules) applyDefaults() {
defaults := DefaultDetectionRules()
r.applyWeightDefaults(defaults.Weights)
r.applySaturationDefaults(defaults.Saturation)
r.applyThresholdDefaults(defaults.Thresholds)
r.applyPenaltyDefaults(defaults.Penalties)
r.applyDegradedMemoryDefaults(defaults.DegradedMemory)
}
func (r *DetectionRules) applyWeightDefaults(defaults RuleWeights) {
defaultPositive(&r.Weights.Cache, defaults.Cache)
defaultPositive(&r.Weights.Memory, defaults.Memory)
defaultPositive(&r.Weights.IO, defaults.IO)
}
func (r *DetectionRules) applySaturationDefaults(defaults RuleSaturation) {
defaultPositive(&r.Saturation.CacheProduct, defaults.CacheProduct)
defaultPositive(&r.Saturation.PSIMemory, defaults.PSIMemory)
defaultPositive(&r.Saturation.BioLatencyP99Ms, defaults.BioLatencyP99Ms)
defaultPositive(&r.Saturation.PSIIO, defaults.PSIIO)
}
func (r *DetectionRules) applyThresholdDefaults(defaults RuleThresholds) {
defaultPositive(&r.Thresholds.CacheMissRate, defaults.CacheMissRate)
defaultPositive(&r.Thresholds.LLCOccupancy, defaults.LLCOccupancy)
defaultPositive(&r.Thresholds.BioLatencyP99Ms, defaults.BioLatencyP99Ms)
defaultPositive(&r.Thresholds.PSIIO, defaults.PSIIO)
defaultPositive(&r.Thresholds.PSIMemorySome, defaults.PSIMemorySome)
defaultPositive(&r.Thresholds.IOPenaltyP99Ms, defaults.IOPenaltyP99Ms)
defaultPositive(&r.Thresholds.IOPenaltyPSIIO, defaults.IOPenaltyPSIIO)
defaultPositive(&r.Thresholds.MemIOPenaltyMemory, defaults.MemIOPenaltyMemory)
defaultPositive(&r.Thresholds.MemIOPenaltyPSIIO, defaults.MemIOPenaltyPSIIO)
}
func (r *DetectionRules) applyPenaltyDefaults(defaults RulePenalties) {
defaultPositive(&r.Penalties.IOMultiplier, defaults.IOMultiplier)
defaultPositive(&r.Penalties.MemIOMultiplier, defaults.MemIOMultiplier)
}
func (r *DetectionRules) applyDegradedMemoryDefaults(defaults RuleDegradedMemory) {
defaultPositive(&r.DegradedMemory.MemAvailableRatioThreshold, defaults.MemAvailableRatioThreshold)
defaultPositive(&r.DegradedMemory.PageScanRateThreshold, defaults.PageScanRateThreshold)
}
func defaultPositive(value *float64, fallback float64) {
if *value <= 0 {
*value = fallback
}
}
type DefaultRuleEngine struct{}
func (DefaultRuleEngine) Evaluate(metrics NodeMetrics) []Detection {
return DefaultDetectionRules().Evaluate(metrics)
}
func (r DetectionRules) Evaluate(metrics NodeMetrics) []Detection {
var detections []Detection
if metrics.LLCMissRate > r.Thresholds.CacheMissRate && metrics.LLCOccupancy > r.Thresholds.LLCOccupancy {
detections = append(detections, Detection{
Type: "cache-pollution",
Message: fmt.Sprintf("LLC miss rate %.2f > %.2f and LLC occupancy %.2f > %.2f", metrics.LLCMissRate, r.Thresholds.CacheMissRate, metrics.LLCOccupancy, r.Thresholds.LLCOccupancy),
})
}
if metrics.BioLatencyP99Ms > r.Thresholds.BioLatencyP99Ms || metrics.PSIIOAvg10 > r.Thresholds.PSIIO {
if !metrics.PSIAvailable && metrics.BioLatencyP99Ms <= r.Thresholds.BioLatencyP99Ms {
return detections
}
detections = append(detections, Detection{
Type: "io-queue-contention",
Message: fmt.Sprintf("Bio latency P99 %.2fms > %.2fms or PSI I/O some %.2f%% > %.2f%%", metrics.BioLatencyP99Ms, r.Thresholds.BioLatencyP99Ms, metrics.PSIIOAvg10, r.Thresholds.PSIIO),
})
}
if metrics.PSIAvailable && (metrics.PSIMemAvg10 > r.Thresholds.PSIMemorySome || metrics.PSIMemFullAvg10 > r.Thresholds.PSIMemoryFull) {
detections = append(detections, Detection{
Type: "memory-pressure",
Message: fmt.Sprintf("PSI memory some %.2f%%, full %.2f%%", metrics.PSIMemAvg10, metrics.PSIMemFullAvg10),
})
} else if !metrics.PSIAvailable && r.degradedMemoryPressure(metrics) {
detections = append(detections, Detection{
Type: "memory-pressure",
Message: fmt.Sprintf("PSI unavailable; mem available ratio %.2f, page scan rate %.2f pages/s", metrics.MemAvailRatio, metrics.PageScanRate),
})
}
return detections
}
type WeightedCalculator struct{}
func (WeightedCalculator) Compute(metrics NodeMetrics) float64 {
return DefaultDetectionRules().Compute(metrics)
}
func (r DetectionRules) Compute(metrics NodeMetrics) float64 {
cacheIndex := r.cacheIndex(metrics)
cacheWeight := r.Weights.Cache
memWeight := r.Weights.Memory
ioWeight := r.Weights.IO
if !metrics.PSIAvailable {
memIndex := r.degradedMemIndex(metrics)
ioIndex := normalize(metrics.BioLatencyP99Ms, r.Saturation.BioLatencyP99Ms)
return clamp(cacheWeight*cacheIndex + memWeight*memIndex + ioWeight*ioIndex)
}
memIndex := normalize(max(metrics.PSIMemAvg10, metrics.PSIMemFullAvg10), r.Saturation.PSIMemory)
ioIndex := max(normalize(metrics.BioLatencyP99Ms, r.Saturation.BioLatencyP99Ms), normalize(metrics.PSIIOAvg10, r.Saturation.PSIIO))
if metrics.PSIIOAvg10 > r.Thresholds.IOPenaltyPSIIO && metrics.BioLatencyP99Ms > r.Thresholds.IOPenaltyP99Ms {
ioWeight *= r.Penalties.IOMultiplier
}
if metrics.PSIMemAvg10 > r.Thresholds.MemIOPenaltyMemory && metrics.PSIIOAvg10 > r.Thresholds.MemIOPenaltyPSIIO {
memWeight *= r.Penalties.MemIOMultiplier
ioWeight *= r.Penalties.MemIOMultiplier
}
return clamp(cacheWeight*cacheIndex + memWeight*memIndex + ioWeight*ioIndex)
}
func (r DetectionRules) cacheIndex(metrics NodeMetrics) float64 {
return normalize(metrics.LLCMissRate*metrics.LLCOccupancy, r.Saturation.CacheProduct)
}
func (r DetectionRules) degradedMemIndex(metrics NodeMetrics) float64 {
availableFactor := normalize(1.0-metrics.MemAvailRatio, r.DegradedMemory.MemAvailableRatioThreshold)
reclaimFactor := normalize(metrics.PageScanRate, r.DegradedMemory.PageScanRateThreshold)
return clamp(0.6*availableFactor + 0.4*reclaimFactor)
}
func (r DetectionRules) degradedMemoryPressure(metrics NodeMetrics) bool {
usedRatio := 1.0 - metrics.MemAvailRatio
return usedRatio >= r.DegradedMemory.MemAvailableRatioThreshold ||
metrics.PageScanRate >= r.DegradedMemory.PageScanRateThreshold
}
func normalize(value, saturation float64) float64 {
if saturation <= 0 {
return 0
}
return clamp(value / saturation)
}
func clamp(value float64) float64 {
if value < 0 {
return 0
}
if value > 1 {
return 1
}
return value
}
func max(a, b float64) float64 {
if a > b {
return a
}
return b
}
