mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2026-09-07 06:17:15 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  dependabot[bot]andGitHub
|
48b37979bd | ||
 Michael Panzlaff
|
6df8da8fe9 | ||
|
Michael Panzlaff
|
39cae9c79a |
@@ -46,6 +46,7 @@ const (
|
||||
LIKWID_LIB_DL_FLAGS = dl.RTLD_LAZY | dl.RTLD_GLOBAL
|
||||
LIKWID_DEF_ACCESSMODE = "direct"
|
||||
LIKWID_DEF_LOCKFILE = "/var/run/likwid.lock"
|
||||
LIKWID_DEF_WARMUP_DEL = "0s"
|
||||
)
|
||||
|
||||
type LikwidCollectorMetricConfig struct {
|
||||
@@ -83,6 +84,7 @@ type LikwidCollectorConfig struct {
|
||||
DaemonPath string `json:"accessdaemon_path,omitempty"`
|
||||
LibraryPath string `json:"liblikwid_path,omitempty"`
|
||||
LockfilePath string `json:"lockfile_path,omitempty"`
|
||||
WarmupDelay string `json:"warmup_delay"`
|
||||
}
|
||||
|
||||
type LikwidCollector struct {
|
||||
@@ -104,6 +106,7 @@ type LikwidCollector struct {
|
||||
likwidGroups map[C.int]LikwidEventsetConfig
|
||||
lock sync.Mutex
|
||||
measureThread thread.Thread
|
||||
warmupDelay time.Duration
|
||||
}
|
||||
|
||||
type LikwidMetric struct {
|
||||
@@ -206,6 +209,7 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
m.config.AccessMode = LIKWID_DEF_ACCESSMODE
|
||||
m.config.LibraryPath = LIKWID_LIB_NAME
|
||||
m.config.LockfilePath = LIKWID_DEF_LOCKFILE
|
||||
m.config.WarmupDelay = LIKWID_DEF_WARMUP_DEL
|
||||
if len(config) > 0 {
|
||||
d := json.NewDecoder(bytes.NewReader(config))
|
||||
d.DisallowUnknownFields()
|
||||
@@ -242,6 +246,17 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
|
||||
m.cpu2tid[c] = i
|
||||
}
|
||||
|
||||
if len(m.config.WarmupDelay) > 0 {
|
||||
t, err := time.ParseDuration(m.config.WarmupDelay)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s Init(): Cannot parse WarmupDelay: %w", m.name, err)
|
||||
}
|
||||
if t < 0 {
|
||||
return fmt.Errorf("%s Init(): WarmupDelay must not be negative", m.name)
|
||||
}
|
||||
m.warmupDelay = t
|
||||
}
|
||||
|
||||
m.likwidGroups = make(map[C.int]LikwidEventsetConfig)
|
||||
|
||||
// This is for the global metrics computation test
|
||||
@@ -495,6 +510,27 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
|
||||
if ret != 0 {
|
||||
return true, fmt.Errorf("failed to start events '%s', error %d", evset.go_estr, ret)
|
||||
}
|
||||
|
||||
// warmup measuring
|
||||
if m.warmupDelay > 0 {
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
case e := <-watcher.Events:
|
||||
if e.Op != fsnotify.Chmod {
|
||||
ret = C.perfmon_readCounters()
|
||||
}
|
||||
default:
|
||||
ret = C.perfmon_readCounters()
|
||||
}
|
||||
if ret != 0 {
|
||||
return true, fmt.Errorf("failed to read events '%s', error %d", evset.go_estr, ret)
|
||||
}
|
||||
|
||||
time.Sleep(m.warmupDelay)
|
||||
}
|
||||
|
||||
// begin measuring
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
@@ -512,7 +548,7 @@ func (m *LikwidCollector) takeMeasurement(evidx int, evset LikwidEventsetConfig,
|
||||
// Wait
|
||||
time.Sleep(interval)
|
||||
|
||||
// Read counters
|
||||
// end measuring
|
||||
select {
|
||||
case <-sigchan:
|
||||
ret = -1
|
||||
|
||||
@@ -63,6 +63,7 @@ Additional options:
|
||||
- `accessdaemon_path`: Folder of the accessDaemon `likwid-accessD` (like `/usr/local/sbin`)
|
||||
- `liblikwid_path`: Location of `liblikwid.so` including file name like `/usr/local/lib/liblikwid.so`
|
||||
- `lockfile_path`: Location of LIKWID's lock file if multiple tools should access the hardware counters. Default `/var/run/likwid.lock`
|
||||
- `warmup_delay`: Run an additional measurement of the specified length before the actual measurement. This can be used as a workaround for CPU starvation (during high load) in the measurement thread. Default is disabled (i.e. `0s`).
|
||||
|
||||
### Available metric types
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ module github.com/ClusterCockpit/cc-metric-collector
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.12.0
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.13.0
|
||||
github.com/ClusterCockpit/go-rocm-smi v0.4.0
|
||||
github.com/NVIDIA/go-nvml v0.13.3-1
|
||||
github.com/PaesslerAG/gval v1.2.4
|
||||
@@ -18,29 +18,28 @@ require (
|
||||
github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/coder/websocket v1.8.14 // indirect
|
||||
github.com/coder/websocket v1.8.15 // indirect
|
||||
github.com/expr-lang/expr v1.17.8 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/gorilla/mux v1.8.1 // indirect
|
||||
github.com/influxdata/influxdb-client-go/v2 v2.14.0 // indirect
|
||||
github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf // indirect
|
||||
github.com/klauspost/compress v1.18.5 // indirect
|
||||
github.com/klauspost/compress v1.19.2 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/nats-io/nats.go v1.51.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.15 // indirect
|
||||
github.com/nats-io/nats.go v1.53.1 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
github.com/oapi-codegen/runtime v1.4.0 // indirect
|
||||
github.com/prometheus/client_golang v1.23.2 // indirect
|
||||
github.com/oapi-codegen/runtime v1.7.0 // indirect
|
||||
github.com/prometheus/client_golang v1.24.1 // indirect
|
||||
github.com/prometheus/client_model v0.6.2 // indirect
|
||||
github.com/prometheus/common v0.67.5 // indirect
|
||||
github.com/prometheus/procfs v0.20.1 // indirect
|
||||
github.com/prometheus/common v0.70.1 // indirect
|
||||
github.com/prometheus/procfs v0.21.1 // indirect
|
||||
github.com/questdb/go-questdb-client/v4 v4.2.0 // indirect
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect
|
||||
github.com/shopspring/decimal v1.4.0 // indirect
|
||||
github.com/stmcginnis/gofish v0.21.6 // indirect
|
||||
github.com/stmcginnis/gofish v0.24.0 // indirect
|
||||
github.com/tklauser/numcpus v0.12.0 // indirect
|
||||
go.yaml.in/yaml/v2 v2.4.4 // indirect
|
||||
golang.org/x/crypto v0.50.0 // indirect
|
||||
golang.org/x/net v0.53.0 // indirect
|
||||
google.golang.org/protobuf v1.36.11 // indirect
|
||||
golang.org/x/crypto v0.55.0 // indirect
|
||||
golang.org/x/net v0.58.0 // indirect
|
||||
google.golang.org/protobuf v1.36.12 // indirect
|
||||
)
|
||||
|
||||
@@ -2,8 +2,8 @@ dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
|
||||
dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0=
|
||||
github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.12.0 h1:ZbGD68nDniuvzFjJCdyYawpCBrabdSyWOg5FFSyFbjQ=
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.12.0/go.mod h1:ml8xtcYa5WhPM7JDQ+M9/R9ZBxITCR/5xqGJ//GxXJI=
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.13.0 h1:Gev5Bl4xuj+u15oTHNBLgpQeGER56e1FVWuTdBrfWLk=
|
||||
github.com/ClusterCockpit/cc-lib/v2 v2.13.0/go.mod h1:F7SmuvPGDqZucdm4MdMK///4RePgzWhRP0cyWK1Cons=
|
||||
github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0 h1:hIzxgTBWcmCIHtoDKDkSCsKCOCOwUC34sFsbD2wcW0Q=
|
||||
github.com/ClusterCockpit/cc-line-protocol/v2 v2.4.0/go.mod h1:y42qUu+YFmu5fdNuUAS4VbbIKxVjxCvbVqFdpdh8ahY=
|
||||
github.com/ClusterCockpit/go-rocm-smi v0.4.0 h1:3+bEPrSkjEJcOtt+qBUX48ugDVlOFaKUnXHTef2Ve2Q=
|
||||
@@ -20,8 +20,8 @@ github.com/PaesslerAG/gval v1.2.4/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbV
|
||||
github.com/PaesslerAG/jsonpath v0.1.0 h1:gADYeifvlqK3R3i2cR5B4DGgxLXIPb3TRTH1mGi0jPI=
|
||||
github.com/PaesslerAG/jsonpath v0.1.0/go.mod h1:4BzmtoM/PI8fPO4aQGIusjGxGir2BzcV0grWtFzq1Y8=
|
||||
github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op h1:kpBdlEPbRvff0mDD1gk7o9BhI16b9p5yYAXRlidpqJE=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.6.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op h1:p2zFsAzvhIpFya8AIOHIbWf7NGvO34QpLGclyf7nXj8=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.2-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
|
||||
github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
@@ -31,8 +31,8 @@ github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqy
|
||||
github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
|
||||
github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
|
||||
github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
|
||||
github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
|
||||
github.com/containerd/containerd v1.7.12 h1:+KQsnv4VnzyxWcfO9mlxxELaoztsDEjOuCMPAuPqgU0=
|
||||
github.com/containerd/containerd v1.7.12/go.mod h1:/5OMpE1p0ylxtEUGY8kuCYkDRzJm9NO1TFMWjUpdevk=
|
||||
github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
|
||||
@@ -77,10 +77,10 @@ github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1
|
||||
github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937 h1:MHJNQ+p99hFATQm6ORoLmpUCF7ovjwEFshs/NHzAbig=
|
||||
github.com/influxdata/line-protocol-corpus v0.0.0-20210922080147-aa28ccfb8937/go.mod h1:BKR9c0uHSmRgM/se9JhFHtTT7JTO67X23MtKMHtZcpo=
|
||||
github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
|
||||
github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
|
||||
github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
|
||||
github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
|
||||
github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
|
||||
github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
|
||||
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
|
||||
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
|
||||
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
|
||||
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
|
||||
@@ -89,8 +89,8 @@ github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0g
|
||||
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=
|
||||
github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
|
||||
github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
|
||||
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76 h1:KGuD/pM2JpL9FAYvBrnBBeENKZNh6eNtjqytV6TYjnk=
|
||||
github.com/minio/highwayhash v1.0.4-0.20251030100505-070ab1a87a76/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
|
||||
github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
|
||||
github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
|
||||
@@ -101,18 +101,20 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
|
||||
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
|
||||
github.com/nats-io/nats-server/v2 v2.12.7 h1:prQ9cPiWHcnwfT81Wi5lU9LL8TLY+7pxDru6fQYLCQQ=
|
||||
github.com/nats-io/nats-server/v2 v2.12.7/go.mod h1:dOnmkprKMluTmTF7/QHZioxlau3sKHUM/LBPy9AiBPw=
|
||||
github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
|
||||
github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
|
||||
github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
|
||||
github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
|
||||
github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc=
|
||||
github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc=
|
||||
github.com/nats-io/nats-server/v2 v2.14.5 h1:M6yeo/Xb7khi97RSEVELof3DForDqmYza3P4tHCPFWw=
|
||||
github.com/nats-io/nats-server/v2 v2.14.5/go.mod h1:1D3iocrisKvWaD1B/imqarTqmaGrWMqALMLbEDo3v7Q=
|
||||
github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo=
|
||||
github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
github.com/oapi-codegen/runtime v1.4.0 h1:KLOSFOp7UzkbS7Cs1ms6NBEKYr0WmH2wZG0KKbd2er4=
|
||||
github.com/oapi-codegen/runtime v1.4.0/go.mod h1:5sw5fxCDmnOzKNYmkVNF8d34kyUeejJEY8HNT2WaPec=
|
||||
github.com/oapi-codegen/nullable v1.1.0 h1:eAh8JVc5430VtYVnq00Hrbpag9PFRGWLjxR1/3KntMs=
|
||||
github.com/oapi-codegen/nullable v1.1.0/go.mod h1:KUZ3vUzkmEKY90ksAmit2+5juDIhIZhfDl+0PwOQlFY=
|
||||
github.com/oapi-codegen/runtime v1.7.0 h1:t7358VYPvNbWJ9gdAkIK/smVeHpBf6yp8VTsaZsb/7k=
|
||||
github.com/oapi-codegen/runtime v1.7.0/go.mod h1:GwV7hC2hviaMzj+ITfHVRESK5J2W/GefVwIND/bMGvU=
|
||||
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
|
||||
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
|
||||
github.com/opencontainers/image-spec v1.1.0-rc5 h1:Ygwkfw9bpDvs+c9E34SdgGOj41dX/cbdlwvlWt0pnFI=
|
||||
@@ -125,18 +127,16 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b h1:0LFwY6Q3gMACTjAbMZBjXAqTOzOwFaj2Ld6cjeQ7Rig=
|
||||
github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
|
||||
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
|
||||
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
|
||||
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
|
||||
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
|
||||
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
|
||||
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
|
||||
github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTUGI4=
|
||||
github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw=
|
||||
github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc=
|
||||
github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
|
||||
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
|
||||
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
|
||||
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
|
||||
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
|
||||
github.com/questdb/go-questdb-client/v4 v4.2.0 h1:+d0HJwCjUWMj7zmY6qmhoqTJzTyoYKl+LSTYGN0T8T8=
|
||||
github.com/questdb/go-questdb-client/v4 v4.2.0/go.mod h1:/2x93LK1wjM4JX/b5c6q7Yqk22htjWY1lE6p1X8iLbE=
|
||||
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
|
||||
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4=
|
||||
github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY=
|
||||
github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4=
|
||||
@@ -149,8 +149,8 @@ github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+D
|
||||
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
|
||||
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
|
||||
github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
|
||||
github.com/stmcginnis/gofish v0.21.6 h1:jK3TGD6VANaAHKHypVNfD6io2nPrU+6eF8X4qARsTlY=
|
||||
github.com/stmcginnis/gofish v0.21.6/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
|
||||
github.com/stmcginnis/gofish v0.24.0 h1:zaBBFNtdSFH/+lJju29HMDHU3suIR+YhygqoJYxW+2Q=
|
||||
github.com/stmcginnis/gofish v0.24.0/go.mod h1:PzF5i8ecRG9A2ol8XT64npKUunyraJ+7t0kYMpQAtqU=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
|
||||
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
|
||||
@@ -176,14 +176,14 @@ go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
|
||||
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
|
||||
golang.design/x/runtime v0.3.0 h1:8bA+GQfO8A18JXJQlQA56pEc+Sgrbo4cmLzjzmJ58ZI=
|
||||
golang.design/x/runtime v0.3.0/go.mod h1:PjVKQjImLdSrvjIw4FmrnCPq9BGs5PMBHb8i6b2BN9I=
|
||||
golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
|
||||
golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
|
||||
golang.org/x/exp v0.0.0-20231005195138-3e424a577f31 h1:9k5exFQKQglLo+RoP+4zMjOFE14P6+vyR0baDAi0Rcs=
|
||||
golang.org/x/exp v0.0.0-20231005195138-3e424a577f31/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
|
||||
golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M=
|
||||
golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis=
|
||||
golang.org/x/exp v0.0.0-20240404231335-c0f41cb1a7a0 h1:985EYyeCOxTpcgOTJpflJUwOeEz0CQOdPt73OzpE9F8=
|
||||
golang.org/x/exp v0.0.0-20240404231335-c0f41cb1a7a0/go.mod h1:/lliqkxwWAhPjf5oSOIJup2XcqJaw8RGS6k3TGEc7GI=
|
||||
golang.org/x/mod v0.13.0 h1:I/DsJXRlw/8l/0c24sM9yb0T4z9liZTduXvdAWYiysY=
|
||||
golang.org/x/mod v0.13.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
|
||||
golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
|
||||
golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
|
||||
golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
|
||||
golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
|
||||
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
@@ -194,8 +194,8 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 h1:
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97/go.mod h1:v7nGkzlmW8P3n/bKmWBn2WpBjpOEx8Q6gMueudAmKfY=
|
||||
google.golang.org/grpc v1.58.3 h1:BjnpXut1btbtgN/6sp+brB2Kbm2LjNXnidYujAVbSoQ=
|
||||
google.golang.org/grpc v1.58.3/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
|
||||
google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
|
||||
@@ -69,8 +69,8 @@ var metricCacheLanguage = gval.NewLanguage(
|
||||
gval.Function("getNumaCpuList", getCpuListOfNumaDomainFunc),
|
||||
gval.Function("getDieCpuList", getCpuListOfDieFunc),
|
||||
gval.Function("getCoreCpuList", getCpuListOfCoreFunc),
|
||||
gval.Function("getCpuList", getCpuListOfNodeFunc),
|
||||
gval.Function("getCpuListOfType", getCpuListOfTypeFunc),
|
||||
gval.Function("getCpuList", getCpuListOfNode),
|
||||
gval.Function("getCpuListOfType", getCpuListOfType),
|
||||
)
|
||||
|
||||
var language gval.Language = gval.NewLanguage(
|
||||
|
||||
@@ -1,449 +0,0 @@
|
||||
package metricAggregator
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"math"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
|
||||
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
|
||||
|
||||
"github.com/expr-lang/expr"
|
||||
"github.com/expr-lang/expr/vm"
|
||||
)
|
||||
|
||||
type MetricAggregatorExprIntervalConfig struct {
|
||||
Name string `json:"name"` // Metric name for the new metric
|
||||
Function string `json:"function"` // Function to apply on the metric
|
||||
Condition string `json:"if"` // Condition for applying function
|
||||
Tags map[string]string `json:"tags,omitempty"` // Tags for the new metric
|
||||
Meta map[string]string `json:"meta,omitempty"` // Meta information for the new metric
|
||||
ValueType string `json:"value_type,omitempty"`
|
||||
IncludeNumCacheIntervals int `json:"expr_on_num_intervals,omitempty"`
|
||||
exprCond *vm.Program
|
||||
exprFunc *vm.Program
|
||||
}
|
||||
|
||||
type metricAggregatorExpr struct {
|
||||
constants map[string]any
|
||||
output chan lp.CCMessage
|
||||
aggregations []MetricAggregatorExprIntervalConfig
|
||||
}
|
||||
|
||||
var paramMapPool = sync.Pool{
|
||||
New: func() any {
|
||||
return make(map[string]any)
|
||||
},
|
||||
}
|
||||
|
||||
func sanitizeExprString(key string) string {
|
||||
return strings.ReplaceAll(key, "type-id", "typeid")
|
||||
}
|
||||
|
||||
// AddAggregation(name, function, condition string, tags, meta map[string]string) error
|
||||
// DeleteAggregation(name string) error
|
||||
// Init(output chan lp.CCMessage) error
|
||||
// Eval(starttime time.Time, endtime time.Time, metrics []lp.CCMessage)
|
||||
|
||||
func (m *metricAggregatorExpr) Init(output chan lp.CCMessage) error {
|
||||
m.output = output
|
||||
m.constants = make(map[string]any)
|
||||
m.aggregations = make([]MetricAggregatorExprIntervalConfig, 0)
|
||||
return nil
|
||||
}
|
||||
|
||||
func GetParamMap(point lp.CCMessage) map[string]any {
|
||||
params := paramMapPool.Get().(map[string]any)
|
||||
clear(params)
|
||||
|
||||
// Put metric name into params map
|
||||
params["name"] = point.Name()
|
||||
|
||||
// Put full message into params map
|
||||
params["message"] = point
|
||||
params["msg"] = point
|
||||
|
||||
// Put timestamp into params map
|
||||
params["timestamp"] = point.Time().Unix()
|
||||
params["time"] = params["timestamp"]
|
||||
|
||||
// Put fields into params map
|
||||
fields := paramMapPool.Get().(map[string]any)
|
||||
clear(fields)
|
||||
for key, value := range point.Fields() {
|
||||
fields[key] = value
|
||||
switch key {
|
||||
case "value":
|
||||
params["messagetype"] = "metric"
|
||||
params["value"] = value
|
||||
params["metric"] = value
|
||||
case "event":
|
||||
params["messagetype"] = "event"
|
||||
params["event"] = value
|
||||
case "control":
|
||||
params["messagetype"] = "control"
|
||||
params["control"] = value
|
||||
case "log":
|
||||
params["messagetype"] = "log"
|
||||
params["log"] = value
|
||||
default:
|
||||
params["messagetype"] = "unknown"
|
||||
}
|
||||
}
|
||||
params["msgtype"] = params["messagetype"]
|
||||
params["fields"] = fields
|
||||
params["field"] = fields
|
||||
|
||||
// Put tags into params map
|
||||
tags := paramMapPool.Get().(map[string]any)
|
||||
clear(tags)
|
||||
for key, value := range point.Tags() {
|
||||
tags[sanitizeExprString(key)] = value
|
||||
}
|
||||
params["tags"] = tags
|
||||
params["tag"] = tags
|
||||
|
||||
// Put meta information into params map
|
||||
meta := paramMapPool.Get().(map[string]any)
|
||||
clear(meta)
|
||||
for key, value := range point.Meta() {
|
||||
meta[sanitizeExprString(key)] = value
|
||||
}
|
||||
params["meta"] = meta
|
||||
|
||||
return params
|
||||
}
|
||||
|
||||
var baseenv_multi_message = map[string]any{
|
||||
"name": "",
|
||||
"starttime": 1234567890,
|
||||
"endtime": 1234567890,
|
||||
"messages": make([]lp.CCMessage, 0),
|
||||
"Median": medianfunc,
|
||||
"Sum": sumfunc,
|
||||
"Min": minfunc,
|
||||
"Max": maxfunc,
|
||||
"Mean": avgfunc,
|
||||
"Avg": avgfunc,
|
||||
"Match": matchfunc,
|
||||
"getCpuCore": getCpuCoreFunc,
|
||||
"getCpuSocket": getCpuSocketFunc,
|
||||
"getCpuNumaDomain": getCpuNumaDomainFunc,
|
||||
"getCpuDie": getCpuDieFunc,
|
||||
"getCpuListOfCore": getCpuListOfCoreFunc,
|
||||
"getCpuListOfSocket": getCpuListOfSocketFunc,
|
||||
"getCpuListOfNumaDomain": getCpuListOfNumaDomainFunc,
|
||||
"getCpuListOfDie": getCpuListOfDieFunc,
|
||||
"getCpuListOfNode": getCpuListOfNodeFunc,
|
||||
"getCpuListOfType": getCpuListOfTypeFunc,
|
||||
}
|
||||
|
||||
func (m *metricAggregatorExpr) AddAggregationWithType(name, function, condition, valueType string, tags, meta map[string]string) error {
|
||||
|
||||
cond, err := expr.Compile(condition, expr.Env(baseenv_multi_message), expr.AsBool(), expr.AllowUndefinedVariables())
|
||||
if err != nil {
|
||||
err = fmt.Errorf("failed to compile condition for aggregation %s: %s", name, err.Error())
|
||||
return err
|
||||
}
|
||||
var exprOption = expr.AsFloat64()
|
||||
switch valueType {
|
||||
case "int":
|
||||
case "int32":
|
||||
exprOption = expr.AsInt()
|
||||
case "int64":
|
||||
exprOption = expr.AsInt64()
|
||||
case "float32":
|
||||
case "float64":
|
||||
exprOption = expr.AsFloat64()
|
||||
case "bool":
|
||||
exprOption = expr.AsBool()
|
||||
default:
|
||||
err := fmt.Errorf("invalid value type '%s' for aggregation %s", valueType, name)
|
||||
return err
|
||||
}
|
||||
f, err := expr.Compile(function, expr.Env(baseenv_multi_message), exprOption, expr.AllowUndefinedVariables())
|
||||
if err != nil {
|
||||
err = fmt.Errorf("failed to compile function for aggregation %s: %s", name, err.Error())
|
||||
return err
|
||||
}
|
||||
m.aggregations = append(m.aggregations, MetricAggregatorExprIntervalConfig{
|
||||
Name: name,
|
||||
Condition: condition,
|
||||
Function: function,
|
||||
Tags: tags,
|
||||
Meta: meta,
|
||||
ValueType: valueType,
|
||||
exprCond: cond,
|
||||
exprFunc: f,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *metricAggregatorExpr) AddAggregation(name, function, condition string, tags, meta map[string]string) error {
|
||||
cclog.ComponentDebugf("MetricAggregator", "Adding %s", name)
|
||||
err := m.AddAggregationWithType(name, function, condition, "float64", tags, meta)
|
||||
cclog.ComponentDebugf("MetricAggregator", "Adding %s returned %v", name, err)
|
||||
return err
|
||||
}
|
||||
|
||||
func (m *metricAggregatorExpr) Eval(starttime, endtime time.Time, metrics []lp.CCMessage) {
|
||||
cclog.ComponentDebugf("MetricAggregator", "Calculating %d expressions", len(m.aggregations))
|
||||
copy_tags := func(tags map[string]string, metrics []lp.CCMessage) map[string]string {
|
||||
out := make(map[string]string)
|
||||
for key, value := range tags {
|
||||
switch value {
|
||||
case "<copy>":
|
||||
for _, m := range metrics {
|
||||
v, err := m.GetTag(key)
|
||||
if err {
|
||||
out[key] = v
|
||||
}
|
||||
}
|
||||
default:
|
||||
out[key] = value
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
copy_meta := func(meta map[string]string, metrics []lp.CCMessage) map[string]string {
|
||||
out := make(map[string]string)
|
||||
for key, value := range meta {
|
||||
switch value {
|
||||
case "<copy>":
|
||||
for _, m := range metrics {
|
||||
v, err := m.GetMeta(key)
|
||||
if err {
|
||||
out[key] = v
|
||||
}
|
||||
}
|
||||
default:
|
||||
out[key] = value
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
for _, aggr := range m.aggregations {
|
||||
selected_metrics := make([]lp.CCMessage, 0)
|
||||
values := make([]float64, 0)
|
||||
aggr_vars := make(map[string]any)
|
||||
maps.Copy(aggr_vars, baseenv_multi_message)
|
||||
maps.Copy(aggr_vars, m.constants)
|
||||
aggr_vars["starttime"] = starttime
|
||||
aggr_vars["endtime"] = endtime
|
||||
|
||||
for _, met := range metrics {
|
||||
met_vars := make(map[string]any)
|
||||
maps.Copy(met_vars, aggr_vars)
|
||||
met_vars["message"] = met
|
||||
maps.Copy(met_vars, GetParamMap(met))
|
||||
res, err := expr.Run(aggr.exprCond, met_vars)
|
||||
if err != nil || res == false {
|
||||
continue
|
||||
}
|
||||
if value, ok := met.GetField("value"); ok {
|
||||
switch v := value.(type) {
|
||||
case float64:
|
||||
values = append(values, v)
|
||||
case float32:
|
||||
case int:
|
||||
case int8:
|
||||
case int16:
|
||||
case int32:
|
||||
case int64:
|
||||
case uint:
|
||||
case uint8:
|
||||
case uint16:
|
||||
case uint32:
|
||||
case uint64:
|
||||
values = append(values, float64(v))
|
||||
case bool:
|
||||
if v {
|
||||
values = append(values, float64(1))
|
||||
} else {
|
||||
values = append(values, float64(0))
|
||||
}
|
||||
default:
|
||||
cclog.ComponentErrorf("MetricAggregator", "Cannot convert value type for %s", met.ToLineProtocol(nil))
|
||||
continue
|
||||
}
|
||||
selected_metrics = append(selected_metrics, met)
|
||||
}
|
||||
}
|
||||
cclog.ComponentDebugf("MetricAggregator", "Collected %d values from %d metrics", len(values), len(selected_metrics))
|
||||
aggr_vars["values"] = values
|
||||
aggr_vars["messages"] = selected_metrics
|
||||
if len(values) > 0 && len(selected_metrics) > 0 {
|
||||
res, err := expr.Run(aggr.exprFunc, aggr_vars)
|
||||
if err == nil {
|
||||
tags := copy_tags(aggr.Tags, selected_metrics)
|
||||
meta := copy_meta(aggr.Meta, selected_metrics)
|
||||
msg, err := lp.NewMetric(aggr.Name, tags, meta, res, time.Now())
|
||||
if err == nil {
|
||||
cclog.ComponentDebugf("MetricAggregator", "Sending %s", msg.ToLineProtocol(nil))
|
||||
select {
|
||||
case m.output <- msg:
|
||||
default:
|
||||
}
|
||||
|
||||
}
|
||||
} else {
|
||||
cclog.ComponentErrorf("MetricAggregator", "Failed to calculate aggregation with name %s: %s", aggr.Name, err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (c *metricAggregatorExpr) AddConstant(name string, value any) {
|
||||
c.constants[name] = value
|
||||
}
|
||||
|
||||
func (c *metricAggregatorExpr) DelConstant(name string) {
|
||||
delete(c.constants, name)
|
||||
}
|
||||
|
||||
func (c *metricAggregatorExpr) DeleteAggregation(name string) error {
|
||||
i := slices.IndexFunc(
|
||||
c.aggregations,
|
||||
func(agg MetricAggregatorExprIntervalConfig) bool {
|
||||
return agg.Name == name
|
||||
})
|
||||
if i == -1 {
|
||||
return fmt.Errorf("no aggregation for metric name %s", name)
|
||||
}
|
||||
copy(c.aggregations[i:], c.aggregations[i+1:])
|
||||
c.aggregations = c.aggregations[:len(c.aggregations)-1]
|
||||
return nil
|
||||
}
|
||||
|
||||
func NewAggregatorExpr(output chan lp.CCMessage) (MetricAggregator, error) {
|
||||
a := new(metricAggregatorExpr)
|
||||
err := a.Init(output)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a, err
|
||||
}
|
||||
|
||||
var expr_cached map[string]*vm.Program = make(map[string]*vm.Program)
|
||||
var expr_cached_lock sync.Mutex
|
||||
|
||||
var baseenv_message = map[string]any{
|
||||
"name": "",
|
||||
"messagetype": "unknown",
|
||||
"msgtype": "unknown",
|
||||
"tag": map[string]any{
|
||||
"type": "unknown",
|
||||
"typeid": "0",
|
||||
"stype": "unknown",
|
||||
"stypeid": "0",
|
||||
"hostname": "localhost",
|
||||
"cluster": "nocluster",
|
||||
},
|
||||
"tags": map[string]any{
|
||||
"type": "unknown",
|
||||
"typeid": "0",
|
||||
"stype": "unknown",
|
||||
"stypeid": "0",
|
||||
"hostname": "localhost",
|
||||
"cluster": "nocluster",
|
||||
},
|
||||
"meta": map[string]any{
|
||||
"unit": "invalid",
|
||||
"source": "unknown",
|
||||
},
|
||||
"fields": map[string]any{
|
||||
"value": 0,
|
||||
"event": "",
|
||||
"control": "",
|
||||
"log": "",
|
||||
},
|
||||
"field": map[string]any{
|
||||
"value": 0,
|
||||
"event": "",
|
||||
"control": "",
|
||||
"log": "",
|
||||
},
|
||||
"timestamp": 1234567890,
|
||||
"msg": lp.EmptyMessage(),
|
||||
"message": lp.EmptyMessage(),
|
||||
}
|
||||
|
||||
func EvalBoolConditionExpr(condition string, msg lp.CCMessage) (bool, error) {
|
||||
scond := sanitizeExprString(condition)
|
||||
expr_cached_lock.Lock()
|
||||
evaluable, ok := expr_cached[scond]
|
||||
expr_cached_lock.Unlock()
|
||||
if !ok {
|
||||
newcond := strings.ReplaceAll(
|
||||
strings.ReplaceAll(
|
||||
scond, "'", "\""), "%", "\\")
|
||||
var err error
|
||||
evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsBool())
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
expr_cached_lock.Lock()
|
||||
expr_cached[scond] = evaluable
|
||||
expr_cached_lock.Unlock()
|
||||
}
|
||||
vars := GetParamMap(msg)
|
||||
|
||||
res, err := expr.Run(evaluable, vars)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
paramMapPool.Put(vars)
|
||||
return res.(bool), nil
|
||||
}
|
||||
|
||||
func EvalFloat64ConditionExpr(condition string, msg lp.CCMessage) (float64, error) {
|
||||
scond := sanitizeExprString(condition)
|
||||
expr_cached_lock.Lock()
|
||||
evaluable, ok := expr_cached[scond]
|
||||
expr_cached_lock.Unlock()
|
||||
if !ok {
|
||||
newcond := strings.ReplaceAll(
|
||||
strings.ReplaceAll(
|
||||
scond, "'", "\""), "%", "\\")
|
||||
var err error
|
||||
evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsFloat64())
|
||||
if err != nil {
|
||||
return math.NaN(), err
|
||||
}
|
||||
expr_cached_lock.Lock()
|
||||
expr_cached[scond] = evaluable
|
||||
expr_cached_lock.Unlock()
|
||||
}
|
||||
vars := GetParamMap(msg)
|
||||
res, err := expr.Run(evaluable, vars)
|
||||
paramMapPool.Put(vars)
|
||||
return res.(float64), err
|
||||
}
|
||||
|
||||
func EvalFloat64Expression(expression string, values map[string]any) (float64, error) {
|
||||
sexpr := sanitizeExprString(expression)
|
||||
expr_cached_lock.Lock()
|
||||
evaluable, ok := expr_cached[sexpr]
|
||||
expr_cached_lock.Unlock()
|
||||
if !ok {
|
||||
newcond := strings.ReplaceAll(
|
||||
strings.ReplaceAll(
|
||||
sexpr, "'", "\""), "%", "\\")
|
||||
var err error
|
||||
evaluable, err = expr.Compile(newcond, expr.Env(baseenv_message), expr.AsFloat64())
|
||||
if err != nil {
|
||||
return math.NaN(), err
|
||||
}
|
||||
expr_cached_lock.Lock()
|
||||
expr_cached[sexpr] = evaluable
|
||||
expr_cached_lock.Unlock()
|
||||
}
|
||||
res, err := expr.Run(evaluable, values)
|
||||
return res.(float64), err
|
||||
}
|
||||
@@ -1,97 +0,0 @@
|
||||
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
|
||||
// All rights reserved. This file is part of cc-lib.
|
||||
// Use of this source code is governed by a MIT-style
|
||||
// license that can be found in the LICENSE file.
|
||||
// additional authors:
|
||||
// Holger Obermaier (NHR@KIT)
|
||||
|
||||
package metricAggregator
|
||||
|
||||
import (
|
||||
"math"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
|
||||
)
|
||||
|
||||
func GenMetricNoCheck(value any, tags, meta map[string]string) lp.CCMessage {
|
||||
msg, _ := lp.NewMetric("test", tags, meta, value, time.Now())
|
||||
return msg
|
||||
}
|
||||
|
||||
type TestBoolConfig struct {
|
||||
msg lp.CCMessage
|
||||
cond string
|
||||
expected_result bool
|
||||
should_fail bool
|
||||
}
|
||||
|
||||
var testBoolConfig []TestBoolConfig = []TestBoolConfig{
|
||||
{
|
||||
msg: GenMetricNoCheck(1.0, nil, nil),
|
||||
cond: "fields.value == 1",
|
||||
expected_result: true,
|
||||
should_fail: false,
|
||||
},
|
||||
{
|
||||
msg: GenMetricNoCheck(1.0, map[string]string{"hostname": "testhost"}, nil),
|
||||
cond: "tags.hostname == 'testhost'",
|
||||
expected_result: true,
|
||||
should_fail: false,
|
||||
},
|
||||
}
|
||||
|
||||
func TestEvalBoolConditionExprSimple(t *testing.T) {
|
||||
|
||||
for _, test := range testBoolConfig {
|
||||
res, err := EvalBoolConditionExpr(test.cond, test.msg)
|
||||
if err != nil && !test.should_fail {
|
||||
t.Error(err.Error())
|
||||
return
|
||||
}
|
||||
if !test.should_fail && res != test.expected_result {
|
||||
t.Errorf("Condition '%s' evaluated to %v despite expecting %v", test.cond, res, test.expected_result)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type TestFloat64Config struct {
|
||||
msg lp.CCMessage
|
||||
cond string
|
||||
expected_result float64
|
||||
should_fail bool
|
||||
}
|
||||
|
||||
var testFloat64Config []TestFloat64Config = []TestFloat64Config{
|
||||
{
|
||||
msg: GenMetricNoCheck(1.0, nil, nil),
|
||||
cond: "fields.value + 3.14",
|
||||
expected_result: 4.14,
|
||||
should_fail: false,
|
||||
},
|
||||
{
|
||||
msg: GenMetricNoCheck(2.0, nil, nil),
|
||||
cond: "fields.value * 2",
|
||||
expected_result: 4.0,
|
||||
should_fail: false,
|
||||
},
|
||||
}
|
||||
|
||||
const compareFloat64Max = 1e-9
|
||||
|
||||
func TestEvalFloat64ConditionExprSimple(t *testing.T) {
|
||||
|
||||
for _, test := range testFloat64Config {
|
||||
res, err := EvalFloat64ConditionExpr(test.cond, test.msg)
|
||||
if err != nil && !test.should_fail {
|
||||
t.Error(err.Error())
|
||||
return
|
||||
}
|
||||
if !test.should_fail && math.Abs(res-test.expected_result) > compareFloat64Max {
|
||||
t.Errorf("Condition '%s' evaluated to %f despite expecting %f", test.cond, res, test.expected_result)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -336,14 +336,14 @@ func getCpuListOfDieFunc(args any) (any, error) {
|
||||
}
|
||||
|
||||
// wrapper function to get a list of all cpuids of the node
|
||||
func getCpuListOfNodeFunc() (any, error) {
|
||||
func getCpuListOfNode() (any, error) {
|
||||
return topo.HwthreadList(), nil
|
||||
}
|
||||
|
||||
// helper function to get the cpuid list for a CCMetric type tag set (type and type-id)
|
||||
// since there is no access to the metric data in the function, is should be called like
|
||||
// `getCpuListOfType()`
|
||||
func getCpuListOfTypeFunc(args ...any) (any, error) {
|
||||
func getCpuListOfType(args ...any) (any, error) {
|
||||
cpulist := make([]int, 0)
|
||||
switch typ := args[0].(type) {
|
||||
case string:
|
||||
|
||||
@@ -9,8 +9,6 @@ package metricRouter
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -21,107 +19,57 @@ import (
|
||||
mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
|
||||
)
|
||||
|
||||
type ccCache struct {
|
||||
periodIdx int
|
||||
maxPeriods int
|
||||
periods [][]lp.CCMessage
|
||||
periodTimes []struct {
|
||||
starttime time.Time
|
||||
endtime time.Time
|
||||
}
|
||||
}
|
||||
|
||||
type CCCache interface {
|
||||
Init(numPeriods int) error
|
||||
Add(msg lp.CCMessage) error
|
||||
GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage)
|
||||
GetAll() []lp.CCMessage
|
||||
NewPeriod()
|
||||
}
|
||||
|
||||
func (c *ccCache) Init(numPeriods int) error {
|
||||
c.maxPeriods = numPeriods
|
||||
c.periodIdx = 0
|
||||
c.periods = make([][]lp.CCMessage, c.maxPeriods)
|
||||
c.periodTimes = make([]struct {
|
||||
starttime time.Time
|
||||
endtime time.Time
|
||||
}, c.maxPeriods)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ccCache) NewPeriod() {
|
||||
c.periodTimes[c.periodIdx].endtime = time.Now()
|
||||
c.periodIdx = (c.periodIdx + 1) % c.maxPeriods
|
||||
fmt.Printf("New period index %d\n", c.periodIdx)
|
||||
c.periods[c.periodIdx] = c.periods[c.periodIdx][:0]
|
||||
c.periodTimes[c.periodIdx].starttime = time.Now()
|
||||
c.periodTimes[c.periodIdx].endtime = c.periodTimes[c.periodIdx].starttime
|
||||
}
|
||||
|
||||
func (c *ccCache) Add(msg lp.CCMessage) error {
|
||||
c.periods[c.periodIdx] = append(c.periods[c.periodIdx], msg)
|
||||
c.periodTimes[c.periodIdx].endtime = msg.Time()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *ccCache) GetPeriod(offset int) (time.Time, time.Time, []lp.CCMessage) {
|
||||
if offset > c.maxPeriods {
|
||||
offset = offset % c.maxPeriods
|
||||
}
|
||||
poff := int(math.Abs(float64(c.periodIdx - offset)))
|
||||
out := make([]lp.CCMessage, 0, len(c.periods[poff%c.maxPeriods]))
|
||||
out = append(out, c.periods[poff%c.maxPeriods]...)
|
||||
|
||||
return c.periodTimes[poff%c.maxPeriods].starttime, c.periodTimes[poff%c.maxPeriods].endtime, out
|
||||
}
|
||||
|
||||
func (c *ccCache) GetAll() []lp.CCMessage {
|
||||
out := make([]lp.CCMessage, 0)
|
||||
for _, data := range c.periods {
|
||||
out = append(out, data...)
|
||||
}
|
||||
return out
|
||||
type metricCachePeriod struct {
|
||||
startstamp time.Time
|
||||
stopstamp time.Time
|
||||
numMetrics int
|
||||
sizeMetrics int
|
||||
metrics []lp.CCMessage
|
||||
}
|
||||
|
||||
// Metric cache data structure
|
||||
type metricCache struct {
|
||||
cache CCCache
|
||||
numPeriods int
|
||||
curPeriod int
|
||||
lock sync.Mutex
|
||||
intervals []*metricCachePeriod
|
||||
wg *sync.WaitGroup
|
||||
ticker mct.MultiChanTicker
|
||||
tickchan chan time.Time
|
||||
done chan bool
|
||||
output chan lp.CCMessage
|
||||
aggEngine agg.MetricAggregator
|
||||
numPeriods int
|
||||
started bool
|
||||
}
|
||||
|
||||
type MetricCache interface {
|
||||
Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error
|
||||
Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error
|
||||
Start()
|
||||
Add(metric lp.CCMessage)
|
||||
GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage)
|
||||
AddAggregation(name, function, condition string, tags, meta map[string]string) error
|
||||
DeleteAggregation(name string) error
|
||||
Close()
|
||||
}
|
||||
|
||||
func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, interval time.Duration, numPeriods int) error {
|
||||
func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) error {
|
||||
var err error
|
||||
c.done = make(chan bool)
|
||||
c.wg = wg
|
||||
c.ticker = ticker
|
||||
c.numPeriods = numPeriods
|
||||
c.started = false
|
||||
c.cache = new(ccCache)
|
||||
c.output = output
|
||||
|
||||
err = c.cache.Init(numPeriods)
|
||||
if err != nil {
|
||||
return fmt.Errorf("MetricCache: failed to create cache: %w", err)
|
||||
c.intervals = make([]*metricCachePeriod, 0)
|
||||
for i := 0; i < c.numPeriods+1; i++ {
|
||||
p := new(metricCachePeriod)
|
||||
p.numMetrics = 0
|
||||
p.sizeMetrics = 0
|
||||
p.metrics = make([]lp.CCMessage, 0)
|
||||
c.intervals = append(c.intervals, p)
|
||||
}
|
||||
|
||||
c.aggEngine, err = agg.NewAggregatorExpr(c.output)
|
||||
// Create a new aggregation engine. No separate goroutine at the moment
|
||||
// The code is executed by the MetricCache goroutine
|
||||
c.aggEngine, err = agg.NewAggregator(c.output)
|
||||
if err != nil {
|
||||
return fmt.Errorf("MetricCache: failed to create aggregator: %w", err)
|
||||
}
|
||||
@@ -133,58 +81,68 @@ func (c *metricCache) Init(output chan lp.CCMessage, ticker mct.MultiChanTicker,
|
||||
func (c *metricCache) Start() {
|
||||
c.tickchan = make(chan time.Time)
|
||||
c.ticker.AddChannel(c.tickchan)
|
||||
// Router cache is done
|
||||
done := func() {
|
||||
cclog.ComponentDebug("MetricCache", "DONE")
|
||||
close(c.done)
|
||||
}
|
||||
|
||||
c.wg.Add(1)
|
||||
go func() {
|
||||
// Rotate cache interval
|
||||
rotate := func(timestamp time.Time) int {
|
||||
oldPeriod := c.curPeriod
|
||||
c.curPeriod = oldPeriod + 1
|
||||
if c.curPeriod >= c.numPeriods {
|
||||
c.curPeriod = 0
|
||||
}
|
||||
c.intervals[oldPeriod].numMetrics = 0
|
||||
c.intervals[oldPeriod].stopstamp = timestamp
|
||||
c.intervals[c.curPeriod].startstamp = timestamp
|
||||
c.intervals[c.curPeriod].stopstamp = timestamp
|
||||
return oldPeriod
|
||||
}
|
||||
|
||||
c.wg.Go(func() {
|
||||
for {
|
||||
select {
|
||||
case <-c.done:
|
||||
c.wg.Done()
|
||||
close(c.done)
|
||||
cclog.ComponentDebug("MetricCache", "DONE")
|
||||
|
||||
done()
|
||||
return
|
||||
case tick := <-c.tickchan:
|
||||
cclog.ComponentDebug("MetricCache", "Tick", tick)
|
||||
allmetrics := c.cache.GetAll()
|
||||
c.cache.NewPeriod()
|
||||
mintime := tick
|
||||
maxtime := mintime.AddDate(-1, 0, 0)
|
||||
|
||||
for _, metric := range allmetrics {
|
||||
if metric.Time().Before(mintime) {
|
||||
mintime = metric.Time()
|
||||
}
|
||||
if metric.Time().After(maxtime) {
|
||||
maxtime = metric.Time()
|
||||
}
|
||||
}
|
||||
if len(allmetrics) > 0 {
|
||||
cclog.ComponentDebugf("MetricCache", "Evaluate %d metrics from %v to %v", len(allmetrics), mintime.UnixNano(), maxtime.UnixNano())
|
||||
c.wg.Go(func() {
|
||||
c.aggEngine.Eval(mintime, maxtime, allmetrics)
|
||||
})
|
||||
c.lock.Lock()
|
||||
old := rotate(tick)
|
||||
// Get the last period and evaluate aggregation metrics
|
||||
starttime, endtime, metrics := c.GetPeriod(old)
|
||||
c.lock.Unlock()
|
||||
if len(metrics) > 0 {
|
||||
c.aggEngine.Eval(starttime, endtime, metrics)
|
||||
} else {
|
||||
// This message is also printed in the first interval after startup
|
||||
cclog.ComponentDebug("MetricCache", "EMPTY INTERVAL?")
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
cclog.ComponentDebug("MetricCache", "STARTED")
|
||||
})
|
||||
cclog.ComponentDebug("MetricCache", "START")
|
||||
}
|
||||
|
||||
// Add a metric to the cache. The interval is defined by the global timer (rotate() in Start())
|
||||
// The intervals list is used as round-robin buffer and the metric list grows dynamically and
|
||||
// to avoid reallocations
|
||||
func (c *metricCache) Add(metric lp.CCMessage) {
|
||||
err := c.cache.Add(metric)
|
||||
if err != nil {
|
||||
s := metric.ToLineProtocol(nil)
|
||||
s = strings.TrimSpace(s)
|
||||
cclog.ComponentErrorf("MetricCache", "Failed to add metric %s", s)
|
||||
if c.curPeriod >= 0 && c.curPeriod < c.numPeriods {
|
||||
c.lock.Lock()
|
||||
p := c.intervals[c.curPeriod]
|
||||
if p.numMetrics < p.sizeMetrics {
|
||||
p.metrics[p.numMetrics] = metric
|
||||
p.numMetrics++
|
||||
p.stopstamp = metric.Time()
|
||||
} else {
|
||||
p.metrics = append(p.metrics, metric)
|
||||
p.numMetrics++
|
||||
p.sizeMetrics++
|
||||
p.stopstamp = metric.Time()
|
||||
}
|
||||
c.lock.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -196,20 +154,40 @@ func (c *metricCache) DeleteAggregation(name string) error {
|
||||
return c.aggEngine.DeleteAggregation(name)
|
||||
}
|
||||
|
||||
// Get all metrics of a interval. The index is the difference to the current interval, so index=0
|
||||
// is the current one, index=1 the last interval and so on. Returns and empty array if a wrong index
|
||||
// is given (negative index, index larger than configured number of total intervals, ...)
|
||||
func (c *metricCache) GetPeriod(index int) (time.Time, time.Time, []lp.CCMessage) {
|
||||
start := time.Now()
|
||||
stop := time.Now()
|
||||
var metrics []lp.CCMessage
|
||||
if index >= 0 && index < c.numPeriods {
|
||||
pindex := c.curPeriod - index
|
||||
if pindex < 0 {
|
||||
pindex = c.numPeriods - pindex
|
||||
}
|
||||
if pindex >= 0 && pindex < c.numPeriods {
|
||||
start = c.intervals[pindex].startstamp
|
||||
stop = c.intervals[pindex].stopstamp
|
||||
metrics = c.intervals[pindex].metrics
|
||||
} else {
|
||||
metrics = make([]lp.CCMessage, 0)
|
||||
}
|
||||
} else {
|
||||
metrics = make([]lp.CCMessage, 0)
|
||||
}
|
||||
return start, stop, metrics
|
||||
}
|
||||
|
||||
// Close finishes / stops the metric cache
|
||||
func (c *metricCache) Close() {
|
||||
cclog.ComponentDebug("MetricCache", "CLOSE")
|
||||
if c.started {
|
||||
c.done <- true
|
||||
c.wg.Wait()
|
||||
|
||||
}
|
||||
cclog.ComponentDebug("MetricCache", "CLOSED")
|
||||
c.done <- true
|
||||
}
|
||||
|
||||
func NewCache(output chan lp.CCMessage, ticker mct.MultiChanTicker, wg *sync.WaitGroup, numPeriods int) (MetricCache, error) {
|
||||
c := new(metricCache)
|
||||
err := c.Init(output, ticker, wg, ticker.GetDuration(), numPeriods)
|
||||
err := c.Init(output, ticker, wg, numPeriods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
|
||||
// All rights reserved. This file is part of cc-lib.
|
||||
// Use of this source code is governed by a MIT-style
|
||||
// license that can be found in the LICENSE file.
|
||||
// additional authors:
|
||||
// Holger Obermaier (NHR@KIT)
|
||||
|
||||
package metricRouter
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
|
||||
mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
|
||||
)
|
||||
|
||||
func TestCache(t *testing.T) {
|
||||
output := make(chan lp.CCMessage, 2000)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
tickTime := time.Second
|
||||
testChan := make(chan time.Time)
|
||||
ticker := mct.NewTicker(tickTime)
|
||||
ticker.AddChannel(testChan)
|
||||
maxIntervals := 100
|
||||
|
||||
c, err := NewCache(output, ticker, &wg, 4)
|
||||
if err != nil {
|
||||
t.Errorf("failed to create new cache: %s", err.Error())
|
||||
return
|
||||
}
|
||||
c.Start()
|
||||
defer c.Close()
|
||||
err = c.AddAggregation("ps_input_power", "Avg(values)", "name == 'ps1_input_power' || name == 'ps2_input_power' || name == 'ps3_input_power'", map[string]string{"hostname": "<copy>", "type": "<copy>"}, map[string]string{"unit": "<copy>"})
|
||||
if err != nil {
|
||||
t.Errorf("failed to add aggregation for %s", "ps_input_power")
|
||||
return
|
||||
}
|
||||
|
||||
for range maxIntervals {
|
||||
timestamp := <-testChan
|
||||
raw_metrics := []string{
|
||||
fmt.Sprintf("ps1_input_power,type=node,hostname=myhost,unit=W value=696.0 %d", timestamp.UnixNano()),
|
||||
fmt.Sprintf("ps2_input_power,type=node,hostname=myhost,unit=W value=732.0 %d", timestamp.UnixNano()),
|
||||
fmt.Sprintf("ps3_input_power,type=node,hostname=myhost,unit=W value=720.0 %d", timestamp.UnixNano()),
|
||||
fmt.Sprintf("cpu_load,type=hwthread,type-id=0,hostname=myhost value=45 %d", timestamp.UnixNano()),
|
||||
}
|
||||
metrics, err := lp.FromBytes([]byte(strings.Join(raw_metrics, "\n")))
|
||||
if err != nil {
|
||||
t.Errorf("failed to generate metrics: %s", err.Error())
|
||||
}
|
||||
for _, m := range metrics {
|
||||
c.Add(m)
|
||||
}
|
||||
|
||||
if len(output) > 0 {
|
||||
for i := 0; i < len(output); i++ {
|
||||
m := <-output
|
||||
t.Log(m.ToLineProtocol(nil))
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -35,19 +35,19 @@ type metricRouterTagConfig struct {
|
||||
|
||||
// Metric router configuration
|
||||
type metricRouterConfig struct {
|
||||
HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
|
||||
AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
|
||||
DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
|
||||
IntervalAgg []agg.MetricAggregatorExprIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
|
||||
DropMetrics []string `json:"drop_metrics"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if
|
||||
DropMetricsIf []string `json:"drop_metrics_if"` // List of evaluatable terms to drop metrics
|
||||
RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value
|
||||
IntervalStamp bool `json:"interval_timestamp"` // Update timestamp periodically by ticker each interval?
|
||||
NumCacheIntervals int `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
|
||||
MaxForward int `json:"max_forward"` // Number of maximal forwarded metrics at one select
|
||||
NormalizeUnits bool `json:"normalize_units"` // Check unit meta flag and normalize it using cc-units
|
||||
ChangeUnitPrefix map[string]string `json:"change_unit_prefix"` // Add prefix that should be applied to the metrics
|
||||
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
|
||||
HostnameTagName string `json:"hostname_tag"` // Key name used when adding the hostname to a metric (default 'hostname')
|
||||
AddTags []metricRouterTagConfig `json:"add_tags"` // List of tags that are added when the condition is met
|
||||
DelTags []metricRouterTagConfig `json:"delete_tags"` // List of tags that are removed when the condition is met
|
||||
IntervalAgg []agg.MetricAggregatorIntervalConfig `json:"interval_aggregates"` // List of aggregation function processed at the end of an interval
|
||||
DropMetrics []string `json:"drop_metrics"` // List of metric names to drop. For fine-grained dropping use drop_metrics_if
|
||||
DropMetricsIf []string `json:"drop_metrics_if"` // List of evaluatable terms to drop metrics
|
||||
RenameMetrics map[string]string `json:"rename_metrics"` // Map to rename metric name from key to value
|
||||
IntervalStamp bool `json:"interval_timestamp"` // Update timestamp periodically by ticker each interval?
|
||||
NumCacheIntervals int `json:"num_cache_intervals"` // Number of intervals of cached metrics for evaluation
|
||||
MaxForward int `json:"max_forward"` // Number of maximal forwarded metrics at one select
|
||||
NormalizeUnits bool `json:"normalize_units"` // Check unit meta flag and normalize it using cc-units
|
||||
ChangeUnitPrefix map[string]string `json:"change_unit_prefix"` // Add prefix that should be applied to the metrics
|
||||
MessageProcessor json.RawMessage `json:"process_messages,omitempty"`
|
||||
}
|
||||
|
||||
// Metric router data structure
|
||||
@@ -253,7 +253,7 @@ func (r *metricRouter) Start() {
|
||||
}
|
||||
// even if the metric is dropped, it is stored in the cache for
|
||||
// aggregations
|
||||
if r.config.NumCacheIntervals > 0 && m != nil {
|
||||
if r.config.NumCacheIntervals > 0 {
|
||||
r.cache.Add(m)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,20 +17,17 @@ type multiChanTicker struct {
|
||||
ticker *time.Ticker
|
||||
channels []chan time.Time
|
||||
done chan bool
|
||||
duration time.Duration
|
||||
}
|
||||
|
||||
type MultiChanTicker interface {
|
||||
Init(duration time.Duration)
|
||||
AddChannel(channel chan time.Time)
|
||||
GetDuration() time.Duration
|
||||
Close()
|
||||
}
|
||||
|
||||
func (t *multiChanTicker) Init(duration time.Duration) {
|
||||
t.ticker = time.NewTicker(duration)
|
||||
t.done = make(chan bool)
|
||||
t.duration = duration
|
||||
go func() {
|
||||
done := func() {
|
||||
close(t.done)
|
||||
@@ -56,10 +53,6 @@ func (t *multiChanTicker) Init(duration time.Duration) {
|
||||
}()
|
||||
}
|
||||
|
||||
func (t *multiChanTicker) GetDuration() time.Duration {
|
||||
return t.duration
|
||||
}
|
||||
|
||||
func (t *multiChanTicker) AddChannel(channel chan time.Time) {
|
||||
t.channels = append(t.channels, channel)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user