Initial version of MQTT sink

Merge branch 'develop' of github.com:ClusterCockpit/cc-metric-collector into develop
Remove stray error check
2025-07-20 20:01:40 +02:00 · 2024-04-19 17:30:18 +02:00 · 2024-04-10 19:57:54 +02:00 · 2024-04-10 19:57:46 +02:00 · 2024-03-06 15:14:44 +01:00 · 2024-03-06 14:59:47 +01:00
10 changed files with 760 additions and 185 deletions
--- a/.github/workflows/Release.yml
+++ b/.github/workflows/Release.yml
@@ -45,10 +45,10 @@ jobs:
    - name: Install build dependencies
      run: |
          dnf --assumeyes install \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.20.6-2.module_el8+658+f14b2092.noarch.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.20.6-1.module_el8+602+8bb8a8d6.x86_64.rpm
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm

    - name: RPM build MetricCollector
      id: rpmbuild
@@ -115,10 +115,10 @@ jobs:
    - name: Install build dependencies
      run: |
          dnf --assumeyes --disableplugin=subscription-manager install \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.20.6-2.module_el8+658+f14b2092.noarch.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.20.6-1.module_el8+602+8bb8a8d6.x86_64.rpm
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm

    - name: RPM build MetricCollector
      id: rpmbuild
--- a/.github/workflows/runonce.yml
+++ b/.github/workflows/runonce.yml
@@ -10,32 +10,6 @@ on:

 jobs:

-  #
-  # Job build-1-20
-  # Build on latest Ubuntu using golang version 1.20
-  #
-  build-1-20:
-    runs-on: ubuntu-latest
-    steps:
-    # See: https://github.com/marketplace/actions/checkout
-    # Checkout git repository and submodules
-    - name: Checkout
-      uses: actions/checkout@v4
-      with:
-        submodules: recursive
-
-    # See: https://github.com/marketplace/actions/setup-go-environment
-    - name: Setup Golang
-      uses: actions/setup-go@v4
-      with:
-        go-version: '1.20'
-
-    - name: Build MetricCollector
-      run: make
-
-    - name: Run MetricCollector once
-      run: ./cc-metric-collector --once --config .github/ci-config.json
-
  #
  # Job build-1-21
  # Build on latest Ubuntu using golang version 1.21
@@ -62,6 +36,32 @@ jobs:
    - name: Run MetricCollector once
      run: ./cc-metric-collector --once --config .github/ci-config.json

+  #
+  # Job build-1-22
+  # Build on latest Ubuntu using golang version 1.22
+  #
+  build-1-22:
+    runs-on: ubuntu-latest
+    steps:
+    # See: https://github.com/marketplace/actions/checkout
+    # Checkout git repository and submodules
+    - name: Checkout
+      uses: actions/checkout@v4
+      with:
+        submodules: recursive
+
+    # See: https://github.com/marketplace/actions/setup-go-environment
+    - name: Setup Golang
+      uses: actions/setup-go@v4
+      with:
+        go-version: '1.22'
+
+    - name: Build MetricCollector
+      run: make
+
+    - name: Run MetricCollector once
+      run: ./cc-metric-collector --once --config .github/ci-config.json
+
  #
  # Build on AlmaLinux 8 using go-toolset
  #
@@ -92,10 +92,10 @@ jobs:
    - name: Install build dependencies
      run: |
          dnf --assumeyes install \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.20.6-2.module_el8+658+f14b2092.noarch.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.20.6-1.module_el8+602+8bb8a8d6.x86_64.rpm
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm

    - name: RPM build MetricCollector
      id: rpmbuild
@@ -130,10 +130,10 @@ jobs:
    - name: Install build dependencies
      run: |
          dnf --assumeyes --disableplugin=subscription-manager install \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.20.6-2.module_el8+658+f14b2092.x86_64.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.20.6-2.module_el8+658+f14b2092.noarch.rpm \
-              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.20.6-1.module_el8+602+8bb8a8d6.x86_64.rpm
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-bin-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/golang-src-1.21.7-1.module_el8+960+4060efbe.noarch.rpm \
+              http://mirror.centos.org/centos/8-stream/AppStream/x86_64/os/Packages/go-toolset-1.21.7-1.module_el8+960+4060efbe.x86_64.rpm

    - name: RPM build MetricCollector
      id: rpmbuild
--- a/collectors/customCmdMetric.go
+++ b/collectors/customCmdMetric.go
@@ -99,10 +99,7 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 				continue
 			}

-			y := lp.FromInfluxMetric(c)
-			if err == nil {
-				output <- y
-			}
+			output <- lp.FromInfluxMetric(c)
 		}
 	}
 	for _, file := range m.files {
@@ -121,10 +118,7 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 			if skip {
 				continue
 			}
-			y := lp.FromInfluxMetric(f)
-			if err == nil {
-				output <- y
-			}
+			output <- lp.FromInfluxMetric(f)
 		}
 	}
 }
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/ClusterCockpit/cc-metric-collector

-go 1.20
+go 1.21

 require (
 	github.com/ClusterCockpit/cc-units v0.4.0
@@ -12,30 +12,30 @@ require (
 	github.com/influxdata/influxdb-client-go/v2 v2.13.0
 	github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf
 	github.com/influxdata/line-protocol/v2 v2.2.1
-	github.com/nats-io/nats.go v1.32.0
-	github.com/prometheus/client_golang v1.18.0
+	github.com/nats-io/nats.go v1.33.1
+	github.com/prometheus/client_golang v1.19.0
 	github.com/stmcginnis/gofish v0.15.0
 	github.com/tklauser/go-sysconf v0.3.13
 	golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1
-	golang.org/x/exp v0.0.0-20240119083558-1b970713d09a
-	golang.org/x/sys v0.16.0
+	golang.org/x/exp v0.0.0-20240222234643-814bf88cf225
+	golang.org/x/sys v0.18.0
 )

 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.2.0 // indirect
-	github.com/google/uuid v1.5.0 // indirect
-	github.com/klauspost/compress v1.17.4 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/klauspost/compress v1.17.7 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oapi-codegen/runtime v1.1.1 // indirect
-	github.com/prometheus/client_model v0.5.0 // indirect
-	github.com/prometheus/common v0.46.0 // indirect
+	github.com/prometheus/client_model v0.6.0 // indirect
+	github.com/prometheus/common v0.49.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
-	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/net v0.20.0 // indirect
-	google.golang.org/protobuf v1.32.0 // indirect
+	golang.org/x/crypto v0.21.0 // indirect
+	golang.org/x/net v0.22.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 )
--- a/go.sum
+++ b/go.sum
@@ -7,7 +7,6 @@ github.com/NVIDIA/go-nvml v0.12.0-2 h1:Sg239yy7jmopu/cuvYauoMj9fOpcGMngxVxxS1EBX
 github.com/NVIDIA/go-nvml v0.12.0-2/go.mod h1:7ruy85eOM73muOc/I37euONSwEyFqZsv5ED9AogD4G0=
 github.com/PaesslerAG/gval v1.2.2 h1:Y7iBzhgE09IGTt5QgGQ2IdaYYYOU134YGHBThD+wm9E=
 github.com/PaesslerAG/gval v1.2.2/go.mod h1:XRFLwvmkTEdYziLdaCeCa5ImcGVrfQbeNUbVR+C6xac=
-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/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
@@ -19,20 +18,16 @@ github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj
 github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/frankban/quicktest v1.11.0/go.mod h1:K+q6oSqb0W0Ininfk863uOk1lMy69l/P6txr3mVT54s=
 github.com/frankban/quicktest v1.11.2/go.mod h1:K+q6oSqb0W0Ininfk863uOk1lMy69l/P6txr3mVT54s=
-github.com/frankban/quicktest v1.13.0 h1:yNZif1OkDfNoDfb9zZa9aXIpejNR4F23Wely0c+Qdqk=
 github.com/frankban/quicktest v1.13.0/go.mod h1:qLE0fzW0VuyUAJgPU19zByoIr0HtCHN/r/VLSOOIySU=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
 github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
-github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
-github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
 github.com/influxdata/influxdb-client-go/v2 v2.13.0 h1:ioBbLmR5NMbAjP4UVA5r9b5xGjpABD7j65pI8kFphDM=
@@ -40,24 +35,20 @@ github.com/influxdata/influxdb-client-go/v2 v2.13.0/go.mod h1:k+spCbt9hcvqvUiz0s
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf h1:7JTmneyiNEwVBOHSjoMxiWAqB992atOeepeFYegn5RU=
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
 github.com/influxdata/line-protocol-corpus v0.0.0-20210519164801-ca6fa5da0184/go.mod h1:03nmhxzZ7Xk2pdG+lmMd7mHDfeVOYFyhOgwO61qWU98=
-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/influxdata/line-protocol/v2 v2.0.0-20210312151457-c52fdecb625a/go.mod h1:6+9Xt5Sq1rWx+glMgxhcg2c0DUaehK+5TDcPZ76GypY=
 github.com/influxdata/line-protocol/v2 v2.1.0/go.mod h1:QKw43hdUBg3GTk2iC3iyCxksNj7PX9aUSeYOYE/ceHY=
 github.com/influxdata/line-protocol/v2 v2.2.1 h1:EAPkqJ9Km4uAxtMRgUubJyqAr6zgWM0dznKMLRauQRE=
 github.com/influxdata/line-protocol/v2 v2.2.1/go.mod h1:DmB3Cnh+3oxmG6LOBIxce4oaL4CPj3OmMPgvauXh+tM=
 github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
-github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
-github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
+github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
-github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
-github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/nats-io/nats.go v1.32.0 h1:Bx9BZS+aXYlxW08k8Gd3yR2s73pV5XSoAQUyp1Kwvp0=
-github.com/nats-io/nats.go v1.32.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nats.go v1.33.1 h1:8TxLZZ/seeEfR97qV0/Bl939tpDnt2Z2fK3HkPypj70=
+github.com/nats-io/nats.go v1.33.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
 github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -65,18 +56,15 @@ github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OS
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/oapi-codegen/runtime v1.1.1 h1:EXLHh0DXIJnWhdRPN2w4MXAzFyE4CskzhNLUmtpMYro=
 github.com/oapi-codegen/runtime v1.1.1/go.mod h1:SK9X900oXmPWilYR5/WKPzt3Kqxn/uS/+lbpREv+eCg=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prometheus/client_golang v1.18.0 h1:HzFfmkOzH5Q8L8G+kSJKUx5dtG87sewO+FoDDqP5Tbk=
-github.com/prometheus/client_golang v1.18.0/go.mod h1:T+GXkCk5wSJyOqMIzVgvvjFDlkOQntgjkJWKrN5txjA=
-github.com/prometheus/client_model v0.5.0 h1:VQw1hfvPvk3Uv6Qf29VrPF32JB6rtbgI6cYPYQjL0Qw=
-github.com/prometheus/client_model v0.5.0/go.mod h1:dTiFglRmd66nLR9Pv9f0mZi7B7fk5Pm3gvsjB5tr+kI=
-github.com/prometheus/common v0.46.0 h1:doXzt5ybi1HBKpsZOL0sSkaNHJJqkyfEWZGGqqScV0Y=
-github.com/prometheus/common v0.46.0/go.mod h1:Tp0qkxpb9Jsg54QMe+EAmqXkSV7Evdy1BTn+g2pa/hQ=
+github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU=
+github.com/prometheus/client_golang v1.19.0/go.mod h1:ZRM9uEAypZakd+q/x7+gmsvXdURP+DABIEIjnmDdp+k=
+github.com/prometheus/client_model v0.6.0 h1:k1v3CzpSRUTrKMppY35TLwPvxHqBu0bYgxZzqGIgaos=
+github.com/prometheus/client_model v0.6.0/go.mod h1:NTQHnmxFpouOD0DpvP4XujX3CdOAGQPoaGhyTchlyt8=
+github.com/prometheus/common v0.49.0 h1:ToNTdK4zSnPVJmh698mGFkDor9wBI/iGaJy5dbH1EgI=
+github.com/prometheus/common v0.49.0/go.mod h1:Kxm+EULxRbUkjGU6WFsQqo3ORzB4tyKvlWFOE9mB2sE=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
-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/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=
 github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
 github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
@@ -88,7 +76,6 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
 github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
@@ -96,21 +83,20 @@ github.com/tklauser/numcpus v0.7.0 h1:yjuerZP127QG9m5Zh/mSO4wqurYil27tHrqwRoRjpr
 github.com/tklauser/numcpus v0.7.0/go.mod h1:bb6dMVcj8A42tSE7i32fsIUCbQNllK5iDguyOZRUzAY=
 golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1 h1:P7S/GeHBAFEZIYp0ePPs2kHXoazz8q2KsyxHyQVGCJg=
 golang.design/x/thread v0.0.0-20210122121316-335e9adffdf1/go.mod h1:9CWpnTUmlQkfdpdutA1nNf4iE5lAVt3QZOu0Z6hahBE=
-golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
-golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
-golang.org/x/exp v0.0.0-20240119083558-1b970713d09a h1:Q8/wZp0KX97QFTc2ywcOE0YRjZPVIx+MXInMzdvQqcA=
-golang.org/x/exp v0.0.0-20240119083558-1b970713d09a/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
-golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
-golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
+golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
+golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
+golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 h1:LfspQV/FYTatPTr/3HzIcmiUFH7PGP+OQ6mgDYo3yuQ=
+golang.org/x/exp v0.0.0-20240222234643-814bf88cf225/go.mod h1:CxmFvTBINI24O/j8iY7H1xHzx2i4OsyguNBmN/uPtqc=
+golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc=
+golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
 golang.org/x/sys v0.0.0-20210122093101-04d7465088b8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
-golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
-google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/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.0-20200615113413-eeeca48fe776/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
--- a/receivers/redfishReceiver.go
+++ b/receivers/redfishReceiver.go
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
 	"strconv"
 	"strings"
@@ -32,10 +33,14 @@ type RedfishReceiverClientConfig struct {

 	doPowerMetric      bool
 	doProcessorMetrics bool
+	doSensors          bool
 	doThermalMetrics   bool

 	skipProcessorMetricsURL map[string]bool

+	// readSensorURLs stores for each chassis ID a list of sensor URLs to read
+	readSensorURLs map[string][]string
+
 	gofish gofish.ClientConfig
 }

@@ -56,7 +61,226 @@ type RedfishReceiver struct {
 	wg   sync.WaitGroup // wait group for redfish receiver
 }

+// deleteEmptyTags removes tags or meta data tags with empty value
+func deleteEmptyTags(tags map[string]string) {
+	maps.DeleteFunc(
+		tags,
+		func(key string, value string) bool {
+			return value == ""
+		},
+	)
+}
+
+// setMetricValue sets the value entry in the fields map
+func setMetricValue(value any) map[string]interface{} {
+	return map[string]interface{}{
+		"value": value,
+	}
+}
+
+// sendMetric sends the metric through the sink channel
+func (r *RedfishReceiver) sendMetric(name string, tags map[string]string, meta map[string]string, value any, timestamp time.Time) {
+
+	deleteEmptyTags(tags)
+	deleteEmptyTags(meta)
+	y, err := lp.New(name, tags, meta, setMetricValue(value), timestamp)
+	if err == nil {
+		r.sink <- y
+	}
+}
+
+// readSensors reads sensors from a redfish device
+// See: https://redfish.dmtf.org/schemas/v1/Sensor.json
+// Redfish URI: /redfish/v1/Chassis/{ChassisId}/Sensors/{SensorId}
+func (r *RedfishReceiver) readSensors(
+	clientConfig *RedfishReceiverClientConfig,
+	chassis *redfish.Chassis) error {
+
+	writeTemperatureSensor := func(sensor *redfish.Sensor) {
+		tags := map[string]string{
+			"hostname": clientConfig.Hostname,
+			"type":     "node",
+			// ChassisType shall indicate the physical form factor for the type of chassis
+			"chassis_typ": string(chassis.ChassisType),
+			// Chassis name
+			"chassis_name": chassis.Name,
+			// ID uniquely identifies the resource
+			"sensor_id": sensor.ID,
+			// The area or device to which this sensor measurement applies
+			"temperature_physical_context": string(sensor.PhysicalContext),
+			// Name
+			"temperature_name": sensor.Name,
+		}
+
+		// Set meta data tags
+		meta := map[string]string{
+			"source": r.name,
+			"group":  "Temperature",
+			"unit":   "degC",
+		}
+
+		r.sendMetric("temperature", tags, meta, sensor.Reading, time.Now())
+	}
+
+	writeFanSpeedSensor := func(sensor *redfish.Sensor) {
+		tags := map[string]string{
+			"hostname": clientConfig.Hostname,
+			"type":     "node",
+			// ChassisType shall indicate the physical form factor for the type of chassis
+			"chassis_typ": string(chassis.ChassisType),
+			// Chassis name
+			"chassis_name": chassis.Name,
+			// ID uniquely identifies the resource
+			"sensor_id": sensor.ID,
+			// The area or device to which this sensor measurement applies
+			"fan_physical_context": string(sensor.PhysicalContext),
+			// Name
+			"fan_name": sensor.Name,
+		}
+
+		// Set meta data tags
+		meta := map[string]string{
+			"source": r.name,
+			"group":  "FanSpeed",
+			"unit":   string(sensor.ReadingUnits),
+		}
+
+		r.sendMetric("fan_speed", tags, meta, sensor.Reading, time.Now())
+	}
+
+	writePowerSensor := func(sensor *redfish.Sensor) {
+		// Set tags
+		tags := map[string]string{
+			"hostname": clientConfig.Hostname,
+			"type":     "node",
+			// ChassisType shall indicate the physical form factor for the type of chassis
+			"chassis_typ": string(chassis.ChassisType),
+			// Chassis name
+			"chassis_name": chassis.Name,
+			// ID uniquely identifies the resource
+			"sensor_id": sensor.ID,
+			// The area or device to which this sensor measurement applies
+			"power_physical_context": string(sensor.PhysicalContext),
+			// Name
+			"power_name": sensor.Name,
+		}
+
+		// Set meta data tags
+		meta := map[string]string{
+			"source": r.name,
+			"group":  "Energy",
+			"unit":   "watts",
+		}
+
+		r.sendMetric("power", tags, meta, sensor.Reading, time.Now())
+	}
+
+	if _, ok := clientConfig.readSensorURLs[chassis.ID]; !ok {
+		// First time run of read sensors for this chassis
+
+		clientConfig.readSensorURLs[chassis.ID] = make([]string, 0)
+
+		// Get sensor information for this chassis
+		sensors, err := chassis.Sensors()
+		if err != nil {
+			return fmt.Errorf("readSensors: chassis.Sensors() failed: %v", err)
+		}
+
+		// Skip empty sensors information
+		if sensors == nil {
+			return nil
+		}
+
+		for _, sensor := range sensors {
+
+			// Skip all sensors which are not in enabled state or which are unhealthy
+			if sensor.Status.State != common.EnabledState || sensor.Status.Health != common.OKHealth {
+				continue
+			}
+
+			// Skip sensors with missing readings units or type
+			if sensor.ReadingUnits == "" || sensor.ReadingType == "" {
+				continue
+			}
+
+			// Power readings
+			if (sensor.ReadingType == redfish.PowerReadingType && sensor.ReadingUnits == "Watts") ||
+				(sensor.ReadingType == redfish.CurrentReadingType && sensor.ReadingUnits == "Watts") {
+				if clientConfig.isExcluded["power"] {
+					continue
+				}
+
+				clientConfig.readSensorURLs[chassis.ID] = append(clientConfig.readSensorURLs[chassis.ID], sensor.ODataID)
+				writePowerSensor(sensor)
+				continue
+			}
+
+			// Fan speed readings
+			if (sensor.ReadingType == redfish.AirFlowReadingType && sensor.ReadingUnits == "RPM") ||
+				(sensor.ReadingType == redfish.AirFlowReadingType && sensor.ReadingUnits == "Percent") {
+				// Skip, when fan_speed metric is excluded
+				if clientConfig.isExcluded["fan_speed"] {
+					continue
+				}
+
+				clientConfig.readSensorURLs[chassis.ID] = append(clientConfig.readSensorURLs[chassis.ID], sensor.ODataID)
+				writeFanSpeedSensor(sensor)
+			}
+
+			// Temperature readings
+			if sensor.ReadingType == redfish.TemperatureReadingType && sensor.ReadingUnits == "C" {
+				if clientConfig.isExcluded["temperature"] {
+					continue
+				}
+
+				clientConfig.readSensorURLs[chassis.ID] = append(clientConfig.readSensorURLs[chassis.ID], sensor.ODataID)
+				writeTemperatureSensor(sensor)
+				continue
+			}
+		}
+	} else {
+
+		common.CollectCollection(
+			func(uri string) {
+				sensor, err := redfish.GetSensor(chassis.GetClient(), uri)
+				if err != nil {
+					cclog.ComponentError(r.name, "redfish.GetSensor() for uri '", uri, "' failed")
+				}
+
+				// Power readings
+				if (sensor.ReadingType == redfish.PowerReadingType && sensor.ReadingUnits == "Watts") ||
+					(sensor.ReadingType == redfish.CurrentReadingType && sensor.ReadingUnits == "Watts") {
+
+					writePowerSensor(sensor)
+					return
+				}
+
+				// Fan speed readings
+				if (sensor.ReadingType == redfish.AirFlowReadingType && sensor.ReadingUnits == "RPM") ||
+					(sensor.ReadingType == redfish.AirFlowReadingType && sensor.ReadingUnits == "Percent") {
+
+					writeFanSpeedSensor(sensor)
+					return
+				}
+
+				// Temperature readings
+				if sensor.ReadingType == redfish.TemperatureReadingType && sensor.ReadingUnits == "C" {
+
+					writeTemperatureSensor(sensor)
+					return
+				}
+			},
+			clientConfig.readSensorURLs[chassis.ID])
+
+	}
+	return nil
+}
+
 // readThermalMetrics reads thermal metrics from a redfish device
+// See: https://redfish.dmtf.org/schemas/v1/Thermal.json
+// Redfish URI: /redfish/v1/Chassis/{ChassisId}/Thermal
+// -> deprecated in favor of the ThermalSubsystem schema
+// -> on Lenovo servers /redfish/v1/Chassis/{ChassisId}/ThermalSubsystem/ThermalMetrics links to /redfish/v1/Chassis/{ChassisId}/Sensors/{SensorId}
 func (r *RedfishReceiver) readThermalMetrics(
 	clientConfig *RedfishReceiverClientConfig,
 	chassis *redfish.Chassis) error {
@@ -106,13 +330,6 @@ func (r *RedfishReceiver) readThermalMetrics(
 			"temperature_name": temperature.Name,
 		}

-		// Delete empty tags
-		for key, value := range tags {
-			if value == "" {
-				delete(tags, key)
-			}
-		}
-
 		// Set meta data tags
 		meta := map[string]string{
 			"source": r.name,
@@ -123,14 +340,7 @@ func (r *RedfishReceiver) readThermalMetrics(
 		// ReadingCelsius shall be the current value of the temperature sensor's reading.
 		value := temperature.ReadingCelsius

-		y, err := lp.New("temperature", tags, meta,
-			map[string]interface{}{
-				"value": value,
-			},
-			timestamp)
-		if err == nil {
-			r.sink <- y
-		}
+		r.sendMetric("temperature", tags, meta, value, timestamp)
 	}

 	for _, fan := range thermal.Fans {
@@ -164,13 +374,6 @@ func (r *RedfishReceiver) readThermalMetrics(
 			"fan_name": fan.Name,
 		}

-		// Delete empty tags
-		for key, value := range tags {
-			if value == "" {
-				delete(tags, key)
-			}
-		}
-
 		// Set meta data tags
 		meta := map[string]string{
 			"source": r.name,
@@ -178,23 +381,16 @@ func (r *RedfishReceiver) readThermalMetrics(
 			"unit":   string(fan.ReadingUnits),
 		}

-		// Reading shall be the current value of the fan sensor's reading
-		value := fan.Reading
-
-		y, err := lp.New("fan_speed", tags, meta,
-			map[string]interface{}{
-				"value": value,
-			},
-			timestamp)
-		if err == nil {
-			r.sink <- y
-		}
+		r.sendMetric("fan_speed", tags, meta, fan.Reading, timestamp)
 	}

 	return nil
 }

 // readPowerMetrics reads power metrics from a redfish device
+// See: https://redfish.dmtf.org/schemas/v1/Power.json
+// Redfish URI: /redfish/v1/Chassis/{ChassisId}/Power
+// -> deprecated in favor of the PowerSubsystem schema
 func (r *RedfishReceiver) readPowerMetrics(
 	clientConfig *RedfishReceiverClientConfig,
 	chassis *redfish.Chassis) error {
@@ -274,13 +470,6 @@ func (r *RedfishReceiver) readPowerMetrics(
 			"power_control_name": pc.Name,
 		}

-		// Delete empty tags
-		for key, value := range tags {
-			if value == "" {
-				delete(tags, key)
-			}
-		}
-
 		// Set meta data tags
 		meta := map[string]string{
 			"source":              r.name,
@@ -289,23 +478,8 @@ func (r *RedfishReceiver) readPowerMetrics(
 			"unit":                "watts",
 		}

-		// Delete empty meta data tags
-		for key, value := range meta {
-			if value == "" {
-				delete(meta, key)
-			}
-		}
-
 		for name, value := range metrics {
-
-			y, err := lp.New(name, tags, meta,
-				map[string]interface{}{
-					"value": value,
-				},
-				timestamp)
-			if err == nil {
-				r.sink <- y
-			}
+			r.sendMetric(name, tags, meta, value, timestamp)
 		}
 	}

@@ -314,6 +488,7 @@ func (r *RedfishReceiver) readPowerMetrics(

 // readProcessorMetrics reads processor metrics from a redfish device
 // See: https://redfish.dmtf.org/schemas/v1/ProcessorMetrics.json
+// Redfish URI: /redfish/v1/Systems/{ComputerSystemId}/Processors/{ProcessorId}/ProcessorMetrics
 func (r *RedfishReceiver) readProcessorMetrics(
 	clientConfig *RedfishReceiverClientConfig,
 	processor *redfish.Processor) error {
@@ -374,13 +549,6 @@ func (r *RedfishReceiver) readProcessorMetrics(
 		"processor_id": processor.ID,
 	}

-	// Delete empty tags
-	for key, value := range tags {
-		if value == "" {
-			delete(tags, key)
-		}
-	}
-
 	// Set meta data tags
 	metaPower := map[string]string{
 		"source": r.name,
@@ -393,14 +561,7 @@ func (r *RedfishReceiver) readProcessorMetrics(
 	if !clientConfig.isExcluded[namePower] &&
 		// Some servers return "ConsumedPowerWatt":65535 instead of "ConsumedPowerWatt":null
 		processorMetrics.ConsumedPowerWatt != 65535 {
-		y, err := lp.New(namePower, tags, metaPower,
-			map[string]interface{}{
-				"value": processorMetrics.ConsumedPowerWatt,
-			},
-			timestamp)
-		if err == nil {
-			r.sink <- y
-		}
+		r.sendMetric(namePower, tags, metaPower, processorMetrics.ConsumedPowerWatt, timestamp)
 	}
 	// Set meta data tags
 	metaThermal := map[string]string{
@@ -412,14 +573,7 @@ func (r *RedfishReceiver) readProcessorMetrics(
 	nameThermal := "temperature"

 	if !clientConfig.isExcluded[nameThermal] {
-		y, err := lp.New(nameThermal, tags, metaThermal,
-			map[string]interface{}{
-				"value": processorMetrics.TemperatureCelsius,
-			},
-			timestamp)
-		if err == nil {
-			r.sink <- y
-		}
+		r.sendMetric(nameThermal, tags, metaThermal, processorMetrics.TemperatureCelsius, timestamp)
 	}
 	return nil
 }
@@ -452,7 +606,8 @@ func (r *RedfishReceiver) readMetrics(clientConfig *RedfishReceiverClientConfig)

 	// Get all chassis managed by this service
 	isChassisListRequired :=
-		clientConfig.doThermalMetrics ||
+		clientConfig.doSensors ||
+			clientConfig.doThermalMetrics ||
 			clientConfig.doPowerMetric
 	var chassisList []*redfish.Chassis
 	if isChassisListRequired {
@@ -472,6 +627,16 @@ func (r *RedfishReceiver) readMetrics(clientConfig *RedfishReceiverClientConfig)
 		}
 	}

+	// Read sensors
+	if clientConfig.doSensors {
+		for _, chassis := range chassisList {
+			err := r.readSensors(clientConfig, chassis)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
 	// read thermal metrics
 	if clientConfig.doThermalMetrics {
 		for _, chassis := range chassisList {
@@ -636,6 +801,7 @@ func NewRedfishReceiver(name string, config json.RawMessage) (Receiver, error) {
 		// Globally disable collection of power, processor or thermal metrics
 		DisablePowerMetrics     bool `json:"disable_power_metrics"`
 		DisableProcessorMetrics bool `json:"disable_processor_metrics"`
+		DisableSensors          bool `json:"disable_sensors"`
 		DisableThermalMetrics   bool `json:"disable_thermal_metrics"`

 		// Globally excluded metrics
@@ -650,6 +816,7 @@ func NewRedfishReceiver(name string, config json.RawMessage) (Receiver, error) {
 			// Per client disable collection of power,processor or thermal metrics
 			DisablePowerMetrics     bool `json:"disable_power_metrics"`
 			DisableProcessorMetrics bool `json:"disable_processor_metrics"`
+			DisableSensors          bool `json:"disable_sensors"`
 			DisableThermalMetrics   bool `json:"disable_thermal_metrics"`

 			// Per client excluded metrics
@@ -766,6 +933,9 @@ func NewRedfishReceiver(name string, config json.RawMessage) (Receiver, error) {
 		doProcessorMetrics :=
 			!(configJSON.DisableProcessorMetrics ||
 				clientConfigJSON.DisableProcessorMetrics)
+		doSensors :=
+			!(configJSON.DisableSensors ||
+				clientConfigJSON.DisableSensors)
 		doThermalMetrics :=
 			!(configJSON.DisableThermalMetrics ||
 				clientConfigJSON.DisableThermalMetrics)
@@ -798,8 +968,10 @@ func NewRedfishReceiver(name string, config json.RawMessage) (Receiver, error) {
 					isExcluded:              isExcluded,
 					doPowerMetric:           doPowerMetric,
 					doProcessorMetrics:      doProcessorMetrics,
+					doSensors:               doSensors,
 					doThermalMetrics:        doThermalMetrics,
 					skipProcessorMetricsURL: make(map[string]bool),
+					readSensorURLs:          map[string][]string{},
 					gofish: gofish.ClientConfig{
 						Username:   username,
 						Password:   password,
--- a/receivers/redfishReceiver.md
+++ b/receivers/redfishReceiver.md
@@ -17,15 +17,17 @@ The Redfish receiver uses the [Redfish (specification)](https://www.dmtf.org/sta
                "host_list": "n[1,2-4]"
            },
            {
-                "host_list": "n5"
-                "disable_power_metrics": true
+                "host_list": "n5",
+                "disable_power_metrics": true,
+                "disable_processor_metrics": true,
+                "disable_thermal_metrics": true
            },
            {
                "host_list": "n6" ],
                "username": "<Username 2>",
                "password": "<Password 2>",
                "endpoint": "https://%h-BMC",
-                "disable_thermal_metrics": true
+                "disable_sensor_metrics": true
            }
        ]
    }
@@ -41,9 +43,18 @@ Global settings:

 Global and per redfish device settings (per redfish device settings overwrite the global settings):

- `disable_power_metrics`: disable collection of power metrics
- `disable_processor_metrics`: disable collection of processor metrics
- `disable_thermal_metrics`: disable collection of thermal metrics
+- `disable_power_metrics`:
+  disable collection of power metrics
+  (`/redfish/v1/Chassis/{ChassisId}/Power`)
+- `disable_processor_metrics`:
+  disable collection of processor metrics
+  (`/redfish/v1/Systems/{ComputerSystemId}/Processors/{ProcessorId}/ProcessorMetrics`)
+- `disable_sensors`:
+  disable collection of fan, power and thermal sensor metrics
+  (`/redfish/v1/Chassis/{ChassisId}/Sensors/{SensorId}`)
+- `disable_thermal_metrics`:
+  disable collection of thermal metrics
+  (`/redfish/v1/Chassis/{ChassisId}/Thermal`)
 - `exclude_metrics`: list of excluded metrics
 - `username`: User name to authenticate with
 - `password`: Password to use for authentication
--- a/sinks/mqttSink.go
+++ b/sinks/mqttSink.go
@@ -0,0 +1,372 @@
+package sinks
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"sync"
+	"time"
+
+	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
+	lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
+	"github.com/go-mqtt/mqtt"
+	influx "github.com/influxdata/line-protocol/v2/lineprotocol"
+	"golang.org/x/exp/slices"
+)
+
+type MqttSinkConfig struct {
+	// defines JSON tags for 'type' and 'meta_as_tags' (string list)
+	// See: metricSink.go
+	defaultSinkConfig
+	// Additional config options, for MqttSink
+	ClientID             string `json:"client_id"`
+	PersistenceDirectory string `json:"persistence_directory,omitempty"`
+	// Maximum number of points sent to server in single request.
+	// Default: 1000
+	BatchSize int `json:"batch_size,omitempty"`
+
+	// Time interval for delayed sending of metrics.
+	// If the buffers are already filled before the end of this interval,
+	// the metrics are sent without further delay.
+	// Default: 1s
+	FlushInterval string `json:"flush_delay,omitempty"`
+	flushDelay    time.Duration
+
+	DialProtocol string `json:"dial_protocol"`
+	Hostname     string `json:"hostname"`
+	Port         int    `json:"port"`
+	PauseTimeout string `json:"pause_timeout"`
+	pauseTimeout time.Duration
+	KeepAlive    uint16 `json:"keep_alive_seconds"`
+	Username     string `json:"username,omitempty"`
+	Password     string `json:"password,omitempty"`
+}
+
+type MqttSink struct {
+	// declares elements 	'name' and 'meta_as_tags' (string to bool map!)
+	sink
+	config MqttSinkConfig // entry point to the MqttSinkConfig
+	// influx line protocol encoder
+	encoder influx.Encoder
+	// number of records stored in the encoder
+	numRecordsInEncoder int
+	// List of tags and meta data tags which should be used as tags
+	extended_tag_list []key_value_pair
+	// Flush() runs in another goroutine and accesses the influx line protocol encoder,
+	// so this encoderLock has to protect the encoder and numRecordsInEncoder
+	encoderLock sync.Mutex
+
+	// timer to run Flush()
+	flushTimer *time.Timer
+	// Lock to assure that only one timer is running at a time
+	timerLock sync.Mutex
+
+	// WaitGroup to ensure only one send operation is running at a time
+	sendWaitGroup sync.WaitGroup
+
+	client     *mqtt.Client
+	mqttconfig mqtt.Config
+}
+
+// Implement functions required for Sink interface
+// Write(...), Flush(), Close()
+// See: metricSink.go
+
+// Code to submit a single CCMetric to the sink
+func (s *MqttSink) Write(m lp.CCMetric) error {
+
+	// Lock for encoder usage
+	s.encoderLock.Lock()
+
+	// Encode measurement name
+	s.encoder.StartLine(m.Name())
+
+	// copy tags and meta data which should be used as tags
+	s.extended_tag_list = s.extended_tag_list[:0]
+	for key, value := range m.Tags() {
+		s.extended_tag_list =
+			append(
+				s.extended_tag_list,
+				key_value_pair{
+					key:   key,
+					value: value,
+				},
+			)
+	}
+	for _, key := range s.config.MetaAsTags {
+		if value, ok := m.GetMeta(key); ok {
+			s.extended_tag_list =
+				append(
+					s.extended_tag_list,
+					key_value_pair{
+						key:   key,
+						value: value,
+					},
+				)
+		}
+	}
+
+	// Encode tags (they musts be in lexical order)
+	slices.SortFunc(
+		s.extended_tag_list,
+		func(a key_value_pair, b key_value_pair) int {
+			if a.key < b.key {
+				return -1
+			}
+			if a.key > b.key {
+				return +1
+			}
+			return 0
+		},
+	)
+	for i := range s.extended_tag_list {
+		s.encoder.AddTag(
+			s.extended_tag_list[i].key,
+			s.extended_tag_list[i].value,
+		)
+	}
+
+	// Encode fields
+	for key, value := range m.Fields() {
+		s.encoder.AddField(key, influx.MustNewValue(value))
+	}
+
+	// Encode time stamp
+	s.encoder.EndLine(m.Time())
+
+	// Check for encoder errors
+	if err := s.encoder.Err(); err != nil {
+		// Unlock encoder usage
+		s.encoderLock.Unlock()
+
+		return fmt.Errorf("encoding failed: %v", err)
+	}
+	s.numRecordsInEncoder++
+
+	if s.config.flushDelay == 0 {
+		// Unlock encoder usage
+		s.encoderLock.Unlock()
+
+		// Directly flush if no flush delay is configured
+		return s.Flush()
+	} else if s.numRecordsInEncoder == s.config.BatchSize {
+		// Unlock encoder usage
+		s.encoderLock.Unlock()
+
+		// Stop flush timer
+		if s.flushTimer != nil {
+			if ok := s.flushTimer.Stop(); ok {
+				cclog.ComponentDebug(s.name, "Write(): Stopped flush timer. Batch size limit reached before flush delay")
+				s.timerLock.Unlock()
+			}
+		}
+
+		// Flush if batch size is reached
+		return s.Flush()
+	} else if s.timerLock.TryLock() {
+
+		// Setup flush timer when flush delay is configured
+		// and no other timer is already running
+		if s.flushTimer != nil {
+
+			// Restarting existing flush timer
+			cclog.ComponentDebug(s.name, "Write(): Restarting flush timer")
+			s.flushTimer.Reset(s.config.flushDelay)
+		} else {
+
+			// Creating and starting flush timer
+			cclog.ComponentDebug(s.name, "Write(): Starting new flush timer")
+			s.flushTimer = time.AfterFunc(
+				s.config.flushDelay,
+				func() {
+					defer s.timerLock.Unlock()
+					cclog.ComponentDebug(s.name, "Starting flush triggered by flush timer")
+					if err := s.Flush(); err != nil {
+						cclog.ComponentError(s.name, "Flush triggered by flush timer: flush failed:", err)
+					}
+				})
+		}
+	}
+
+	// Unlock encoder usage
+	s.encoderLock.Unlock()
+	return nil
+}
+
+// If the sink uses batched sends internally, you can tell to flush its buffers
+func (s *MqttSink) Flush() error {
+
+	// Lock for encoder usage
+	// Own lock for as short as possible: the time it takes to clone the buffer.
+	s.encoderLock.Lock()
+
+	buf := slices.Clone(s.encoder.Bytes())
+	numRecordsInBuf := s.numRecordsInEncoder
+	s.encoder.Reset()
+	s.numRecordsInEncoder = 0
+
+	// Unlock encoder usage
+	s.encoderLock.Unlock()
+
+	if len(buf) == 0 {
+		return nil
+	}
+
+	cclog.ComponentDebug(s.name, "Flush(): Flushing", numRecordsInBuf, "metrics")
+
+	// Asynchron send of encoder metrics
+	s.sendWaitGroup.Add(1)
+	go func() {
+		defer s.sendWaitGroup.Done()
+		//startTime := time.Now()
+		for {
+			exchange, err := s.client.PublishAtLeastOnce(buf, s.config.ClientID)
+			switch {
+			case err == nil:
+				return
+
+			case mqtt.IsDeny(err), errors.Is(err, mqtt.ErrClosed):
+				return
+
+			case errors.Is(err, mqtt.ErrMax):
+				time.Sleep(s.config.pauseTimeout)
+
+			default:
+				time.Sleep(s.config.pauseTimeout)
+				continue
+			}
+
+			for err := range exchange {
+				if errors.Is(err, mqtt.ErrClosed) {
+					return
+				}
+
+			}
+			return
+		}
+	}()
+	return nil
+}
+
+// Close sink: close network connection, close files, close libraries, ...
+func (s *MqttSink) Close() {
+
+	cclog.ComponentDebug(s.name, "CLOSE")
+
+	// Stop existing timer and immediately flush
+	if s.flushTimer != nil {
+		if ok := s.flushTimer.Stop(); ok {
+			s.timerLock.Unlock()
+		}
+	}
+
+	// Flush
+	if err := s.Flush(); err != nil {
+		cclog.ComponentError(s.name, "Close():", "Flush failed:", err)
+	}
+
+	// Wait for send operations to finish
+	s.sendWaitGroup.Wait()
+
+	s.client.Close()
+	s.client = nil
+}
+
+// New function to create a new instance of the sink
+// Initialize the sink by giving it a name and reading in the config JSON
+func NewMqttSink(name string, config json.RawMessage) (Sink, error) {
+	s := new(MqttSink)
+
+	// Set name of sampleSink
+	// The name should be chosen in such a way that different instances of MqttSink can be distinguished
+	s.name = fmt.Sprintf("MqttSink(%s)", name) // Always specify a name here
+
+	// Set defaults in s.config
+	// Allow overwriting these defaults by reading config JSON
+
+	s.config.PauseTimeout = "4s"
+	s.config.pauseTimeout = time.Duration(4) * time.Second
+	s.config.DialProtocol = "tcp"
+	s.config.Hostname = "localhost"
+	s.config.Port = 1883
+
+	// Read in the config JSON
+	if len(config) > 0 {
+		d := json.NewDecoder(bytes.NewReader(config))
+		d.DisallowUnknownFields()
+		if err := d.Decode(&s.config); err != nil {
+			cclog.ComponentError(s.name, "Error reading config:", err.Error())
+			return nil, err
+		}
+	}
+
+	// Create lookup map to use meta infos as tags in the output metric
+	s.meta_as_tags = make(map[string]bool)
+	for _, k := range s.config.MetaAsTags {
+		s.meta_as_tags[k] = true
+	}
+
+	// Check if all required fields in the config are set
+	// E.g. use 'len(s.config.Option) > 0' for string settings
+	if t, err := time.ParseDuration(s.config.PauseTimeout); err == nil {
+		s.config.pauseTimeout = t
+	} else {
+		err := fmt.Errorf("to parse duration for PauseTimeout: %s", s.config.PauseTimeout)
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+	if t, err := time.ParseDuration(s.config.FlushInterval); err == nil {
+		s.config.flushDelay = t
+	} else {
+		err := fmt.Errorf("to parse duration for FlushInterval: %s", s.config.FlushInterval)
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+
+	switch s.config.DialProtocol {
+	case "tcp", "udp":
+	default:
+		err := errors.New("failed to parse dial protocol, allowed: tcp, udp")
+		cclog.ComponentError(s.name, err.Error())
+		return nil, err
+	}
+
+	var persistence mqtt.Persistence
+	if len(s.config.PersistenceDirectory) > 0 {
+		persistence = mqtt.FileSystem(s.config.PersistenceDirectory)
+	} else {
+		tmpdir, err := os.MkdirTemp("", "mqtt")
+		if err == nil {
+			persistence = mqtt.FileSystem(tmpdir)
+		}
+	}
+
+	// Establish connection to the server, library, ...
+	// Check required files exist and lookup path(s) of executable(s)
+
+	dialer := mqtt.NewDialer(s.config.DialProtocol, net.JoinHostPort(s.config.Hostname, fmt.Sprintf("%d", s.config.Port)))
+
+	s.mqttconfig = mqtt.Config{
+		Dialer:       dialer,
+		PauseTimeout: s.config.pauseTimeout,
+		KeepAlive:    uint16(s.config.KeepAlive),
+	}
+	if len(s.config.Username) > 0 {
+		s.mqttconfig.UserName = s.config.Username
+	}
+	if len(s.config.Password) > 0 {
+		s.mqttconfig.Password = []byte(s.config.Password)
+	}
+
+	client, err := mqtt.InitSession(s.config.ClientID, persistence, &s.mqttconfig)
+	if err != nil {
+		return nil, err
+	}
+	s.client = client
+
+	// Return (nil, meaningful error message) in case of errors
+	return s, nil
+}
--- a/sinks/mqttSink.md
+++ b/sinks/mqttSink.md
@@ -0,0 +1,39 @@
+## `mqtt` sink
+
+The `mqtt` sink publishes all metrics into a MQTT network.
+
+### Configuration structure
+
+```json
+{
+  "<name>": {
+    "type": "mqtt",
+    "client_id" : "myid",
+    "persistence_directory": "/tmp",
+    "batch_size": 1000,
+    "flush_delay": "1s",
+    "dial_protocol": "tcp",
+    "host": "dbhost.example.com",
+    "port": 1883,
+    "user": "exampleuser",
+    "password" : "examplepw",
+    "pause_timeout": "1s",
+    "keep_alive_seconds": 10,
+    "meta_as_tags" : [],
+  }
+}
+```
+
+- `type`: makes the sink an `mqtt` sink
+- `client_id`: MQTT clients use a client_id to talk to the broker
+- `persistence_directory`: MQTT stores messages temporarly on disc if the broker is not available. Folder needs to be writable (default: `/tmp`)
+- `pause_timeout`: Waittime when published failed
+- `keep_alive_seconds`: Keep the connection alive for some time. Recommended to be longer than global `interval`.
+- `flush_delay`: Group metrics coming in to a single batch
+- `batch_size`: Maximal batch size. If `batch_size` is reached before the end of `flush_delay`, the metrics are sent without further delay
+- `dial_protocol`: Use `tcp` or `udp` for the MQTT communication
+- `host`: Hostname of the MQTT broker
+- `port`: Port number of the MQTT broker
+- `user`: Username for authentication
+- `password`: Password for authentication
+- `meta_as_tags`: print all meta information as tags in the output (optional)
--- a/sinks/sinkManager.go
+++ b/sinks/sinkManager.go
@@ -21,6 +21,7 @@ var AvailableSinks = map[string]func(name string, config json.RawMessage) (Sink,
 	"influxdb":    NewInfluxSink,
 	"influxasync": NewInfluxAsyncSink,
 	"http":        NewHttpSink,
+	"mqtt":        NewMqttSink,
 }

 // Metric collector manager data structure
Author	SHA1	Message	Date
Thomas Roehl	e1f05acce0	Initial version of MQTT sink	2024-04-19 17:30:18 +02:00
Thomas Roehl	16c796a2b8	Merge branch 'develop' of github.com:ClusterCockpit/cc-metric-collector into develop	2024-04-10 19:57:54 +02:00
Thomas Roehl	b6c4769db3	Remove stray error check	2024-04-10 19:57:46 +02:00
Holger Obermaier	7bbee70c14	Update golang toolchain to 1.21	2024-03-06 15:14:44 +01:00
Holger Obermaier	902f4349b6	Read sensors through redfish	2024-03-06 14:59:47 +01:00