Add derived values for nfsiostat (#132)

* Check creation of CCMessage in NATS receiver

* add derived_values for nfsiostatMetric

---------

Co-authored-by: Thomas Roehl <thomas.roehl@fau.de>
Co-authored-by: exterr2f <Robert.Externbrink@rub.de>
Co-authored-by: Thomas Gruber <Thomas.Roehl@googlemail.com>

collectors/nfsiostatMetric.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
-	lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
+	lp "github.com/ClusterCockpit/cc-lib/ccMessage"
 )
 
 // These are the fields we read from the JSON configuration
@@ -18,17 +18,20 @@ type NfsIOStatCollectorConfig struct {
 	ExcludeMetrics          []string `json:"exclude_metrics,omitempty"`
 	ExcludeFilesystem       []string `json:"exclude_filesystem,omitempty"`
 	UseServerAddressAsSType bool     `json:"use_server_as_stype,omitempty"`
+	SendAbsoluteValues      bool     `json:"send_abs_values"`
+	SendDerivedValues       bool     `json:"send_derived_values"`
 }
 
 // This contains all variables we need during execution and the variables
 // defined by metricCollector (name, init, ...)
 type NfsIOStatCollector struct {
 	metricCollector
-	config NfsIOStatCollectorConfig    // the configuration structure
+	config        NfsIOStatCollectorConfig    // the configuration structure
-	meta   map[string]string           // default meta information
+	meta          map[string]string           // default meta information
-	tags   map[string]string           // default tags
+	tags          map[string]string           // default tags
-	data   map[string]map[string]int64 // data storage for difference calculation
+	data          map[string]map[string]int64 // data storage for difference calculation
-	key    string                      // which device info should be used as subtype ID? 'server' or 'mntpoint', see NfsIOStatCollectorConfig.UseServerAddressAsSType
+	key           string                      // which device info should be used as subtype ID? 'server' or 'mntpoint'
+	lastTimestamp time.Time
 }
 
 var deviceRegex = regexp.MustCompile(`device (?P<server>[^ ]+) mounted on (?P<mntpoint>[^ ]+) with fstype nfs(?P<version>\d*) statvers=[\d\.]+`)
@@ -81,7 +84,6 @@ func (m *NfsIOStatCollector) readNfsiostats() map[string]map[string]int64 {
 							data[current[m.key]][name] = val
 						}
 					}
-
 				}
 				current = nil
 			}
@@ -98,6 +100,9 @@ func (m *NfsIOStatCollector) Init(config json.RawMessage) error {
 	m.meta = map[string]string{"source": m.name, "group": "NFS", "unit": "bytes"}
 	m.tags = map[string]string{"type": "node"}
 	m.config.UseServerAddressAsSType = false
+	// Set default configuration
+	m.config.SendAbsoluteValues = true
+	m.config.SendDerivedValues = false
 	if len(config) > 0 {
 		err = json.Unmarshal(config, &m.config)
 		if err != nil {
@@ -110,12 +115,15 @@ func (m *NfsIOStatCollector) Init(config json.RawMessage) error {
 		m.key = "server"
 	}
 	m.data = m.readNfsiostats()
+	m.lastTimestamp = time.Now()
 	m.init = true
 	return err
 }
 
 func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessage) {
-	timestamp := time.Now()
+	now := time.Now()
+	timeDiff := now.Sub(m.lastTimestamp).Seconds()
+	m.lastTimestamp = now
 
 	// Get the current values for all mountpoints
 	newdata := m.readNfsiostats()
@@ -123,21 +131,30 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
 	for mntpoint, values := range newdata {
 		// Was the mount point already present in the last iteration
 		if old, ok := m.data[mntpoint]; ok {
-			// Calculate the difference of old and new values
+			for name, newVal := range values {
-			for i := range values {
+				if m.config.SendAbsoluteValues {
-				x := values[i] - old[i]
+					msg, err := lp.NewMessage(fmt.Sprintf("nfsio_%s", name), m.tags, m.meta, map[string]interface{}{"value": newVal}, now)
-				y, err := lp.NewMessage(fmt.Sprintf("nfsio_%s", i), m.tags, m.meta, map[string]interface{}{"value": x}, timestamp)
+					if err == nil {
-				if err == nil {
+						msg.AddTag("stype", "filesystem")
-					if strings.HasPrefix(i, "page") {
+						msg.AddTag("stype-id", mntpoint)
-						y.AddMeta("unit", "4K_Pages")
+						output <- msg
 					}
-					y.AddTag("stype", "filesystem")
-					y.AddTag("stype-id", mntpoint)
-					// Send it to output channel
-					output <- y
 				}
-				// Update old to the new value for the next iteration
+				if m.config.SendDerivedValues {
-				old[i] = values[i]
+					rate := float64(newVal-old[name]) / timeDiff
+					msg, err := lp.NewMessage(fmt.Sprintf("nfsio_%s_bw", name), m.tags, m.meta, map[string]interface{}{"value": rate}, now)
+					if err == nil {
+						if strings.HasPrefix(name, "page") {
+							msg.AddMeta("unit", "4K_pages/s")
+						} else {
+							msg.AddMeta("unit", "bytes/sec")
+						}
+						msg.AddTag("stype", "filesystem")
+						msg.AddTag("stype-id", mntpoint)
+						output <- msg
+					}
+				}
+				old[name] = newVal
 			}
 		} else {
 			// First time we see this mount point, store all values
@@ -157,7 +174,6 @@ func (m *NfsIOStatCollector) Read(interval time.Duration, output chan lp.CCMessa
 			m.data[mntpoint] = nil
 		}
 	}
-
 }
 
 func (m *NfsIOStatCollector) Close() {

collectors/nfsiostatMetric.md

@@ -3,16 +3,18 @@
 ```json
   "nfsiostat": {
     "exclude_metrics": [
-      "nfsio_oread"
+      "oread", "pageread"
     ],
-    "exclude_filesystems" : [
+    "exclude_filesystems": [
-        "/mnt",
+      "/mnt"
     ],
-    "use_server_as_stype": false
+    "use_server_as_stype": false,
+    "send_abs_values": false,
+    "send_derived_values": true
   }
 ```
 
-The `nfsiostat` collector reads data from `/proc/self/mountstats` and outputs a handful **node** metrics for each NFS filesystem. If a metric or filesystem is not required, it can be excluded from forwarding it to the sink.
+The `nfsiostat` collector reads data from `/proc/self/mountstats` and outputs a handful **node** metrics for each NFS filesystem. If a metric or filesystem is not required, it can be excluded from forwarding it to the sink. **Note:** When excluding metrics, you must provide the base metric name (e.g. pageread) without the nfsio_ prefix. This exclusion applies to both absolute and derived values.
 
 Metrics:
 * `nfsio_nread`: Bytes transferred by normal `read()` calls
@@ -24,4 +26,9 @@ Metrics:
 * `nfsio_nfsread`: Bytes transferred for reading from the server
 * `nfsio_nfswrite`: Pages transferred by writing to the server
 
-The `nfsiostat` collector adds the mountpoint to the tags as `stype=filesystem,stype-id=<mountpoint>`. If the server address should be used instead of the mountpoint, use the `use_server_as_stype` config setting.
+For each of these, if derived values are enabled, an additional metric is sent with the `_bw` suffix, which represents the rate:
+
+  * For normal byte metrics: `unit=bytes/sec`
+  * For page metrics: `unit=4K_pages/s`
+
+The `nfsiostat` collector adds the mountpoint to the tags as `stype=filesystem,stype-id=<mountpoint>`. If the server address should be used instead of the mountpoint, use the `use_server_as_stype` config setting.