Add stream_stats reset=true support

ldms/python/ldms.pxd

@@ -816,7 +816,7 @@ cdef extern from "ldms.h" nogil:
         pass
     int ldms_stream_stats_level_set(int level)
     int ldms_stream_stats_level_get()
-    ldms_stream_stats_tq_s * ldms_stream_stats_tq_get(const char *match, int is_regex)
+    ldms_stream_stats_tq_s * ldms_stream_stats_tq_get(const char *match, int is_regex, int is_reset)
     void ldms_stream_stats_tq_free(ldms_stream_stats_tq_s *tq)
     char *ldms_stream_stats_tq_to_str(ldms_stream_stats_tq_s *tq)
     char *ldms_stream_stats_str(char *match, int is_regex)

ldms/python/ldms.pyx

@@ -3972,7 +3972,7 @@ def stream_stats_level_set(lvl):
 def stream_stats_level_get():
     return ldms_stream_stats_level_get()
 
-def stream_stats_get(stream_match=None, is_regex=0):
+def stream_stats_get(stream_match=None, is_regex=0, is_reset=0):
     """Get a collection of stats of the streams in this process that match `stream_match`
 
     stream_match(str) - the stream name or a regular expression
@@ -3984,7 +3984,7 @@ def stream_stats_get(stream_match=None, is_regex=0):
     if stream_match:
         stream_match = BYTES(stream_match)
         m = stream_match
-    tq = ldms_stream_stats_tq_get(m, is_regex)
+    tq = ldms_stream_stats_tq_get(m, is_regex, is_reset)
     ret = list()
     if not tq:
         if errno == ENOENT:

ldms/python/ldmsd/ldmsd_communicator.py

@@ -142,7 +142,7 @@
                       'subscribe': {'req_attr': ['name'], 'opt_attr': []},
                       'stream_client_dump': {'req_attr': [], 'opt_attr': []},
                       'stream_status' : {'req_attr': [], 'opt_attr': ['reset']},
-                      'stream_stats' : {'req_attr': [], 'opt_attr': ['regex', 'stream', 'json']},
+                      'stream_stats' : {'req_attr': [], 'opt_attr': ['regex', 'stream', 'json', 'reset']},
                       'stream_client_stats' : {'req_attr': [], 'opt_attr': ['json']},
                       ##### Daemon #####
                       'daemon_status': {'req_attr': [], 'opt_attr': ['thread_stats']},
@@ -1393,19 +1393,22 @@ def stream_status(self, reset = False):
         except Exception as e:
             return errno.ENOTCONN, str(e)
 
-    def stream_stats(self, regex=None, stream=None):
+    def stream_stats(self, regex=None, stream=None, reset=None):
         """
         Dump stream stats
 
         Parameters:
         regex - The regular expression matching the stream names
         stream - The exact match of the stearm name
+        reset - Reset the statistics
         """
         attr_list = []
         if regex:
             attr_list.append(LDMSD_Req_Attr(attr_name='regex', value=regex))
         if stream:
             attr_list.append(LDMSD_Req_Attr(attr_name='stream', value=stream))
+        if reset:
+            attr_list.append(LDMSD_Req_Attr(attr_name='reset', value=reset))
         req = LDMSD_Request(command_id=LDMSD_Request.STREAM_STATS, attrs = attr_list)
         try:
             req.send(self)

ldms/python/ldmsd/ldmsd_controller

@@ -2508,7 +2508,7 @@ class LdmsdCmdParser(cmd.Cmd):
         arg = self.handle_args('stream_stats', arg)
         if not arg:
             return
-        rc, msg = self.comm.stream_stats(arg['regex'], arg['stream'])
+        rc, msg = self.comm.stream_stats(arg['regex'], arg['stream'], arg['reset'])
         if rc:
             emsg = f"stream_stats error: {rc}, msg: {msg}"
             print(emsg)

ldms/src/core/ldms.h

@@ -1309,14 +1309,16 @@ int ldms_stream_stats_level_get();
  *
  * \param match    The stream name or a regular expression.
  * \param is_regex 1 if \c match is a regular expression; otherwise, 0.
+ * \param is_reset 1 means to reset the streams' statistics
  *
  * \retval tq   The collection (tailq) of statistics of the matching entries, or
  * \retval NULL if there is an error. \c errno is also set to describe the error.
  *
  * \note The caller is responsible for freeing the \c rbt and the entries in it.
  *       \c ldms_stream_stats_tq_free() is a helping function for this.
  */
-struct ldms_stream_stats_tq_s * ldms_stream_stats_tq_get(const char *match, int is_regex);
+struct ldms_stream_stats_tq_s *
+ldms_stream_stats_tq_get(const char *match, int is_regex, int is_reset);
 
 /**
  * \brief Free all of the entries in the given \c tq and the \c tq itself.
@@ -1337,12 +1339,13 @@ char *ldms_stream_stats_tq_to_str(struct ldms_stream_stats_tq_s *tq);
  *
  * \param  match    The stream name or a regular expression.
  * \param  is_regex 1 if \c match is a regular expression; otherwise, 0.
+ * \param  is_reset 1 means to reset the streams' statistics
  *
  * \retval str The string describing the stats.
  *
  * \note The caller is responsible for freeing the returned string.
  */
-char *ldms_stream_stats_str(const char *match, int is_regex);
+char *ldms_stream_stats_str(const char *match, int is_regex, int is_reset);
 
 /**
  * Returns a collection of stats of stream clients.

ldms/src/core/ldms_stream.c

@@ -1423,7 +1423,7 @@ void __src_stats_rbt_purge(struct rbt *rbt)
 }
 
 /* copy entries from t0 into t1 */
-int __src_stats_rbt_copy(struct rbt *t0, struct rbt *t1)
+int __src_stats_rbt_copy(struct rbt *t0, struct rbt *t1, int is_reset)
 {
 	struct rbn *rbn;
 	struct ldms_stream_src_stats_s *s0, *s1;
@@ -1436,6 +1436,8 @@ int __src_stats_rbt_copy(struct rbt *t0, struct rbt *t1)
 			goto err_0;
 		}
 		*s1 = *s0;
+		if (is_reset)
+			LDMS_STREAM_COUNTERS_INIT(&s0->rx);
 		rbn_init(&s1->rbn, &s1->src);
 		rbt_ins(t1, &s1->rbn);
 	}
@@ -1458,7 +1460,7 @@ void __stream_stats_free(struct ldms_stream_stats_s *ss)
 }
 
 /* readlock already taken */
-struct ldms_stream_stats_s * __stream_get_stats(struct ldms_stream_s *s)
+struct ldms_stream_stats_s * __stream_get_stats(struct ldms_stream_s *s, int is_reset)
 {
 	/* s->name_len already includes '\0' */
 	struct ldms_stream_stats_s *ss;
@@ -1476,7 +1478,7 @@ struct ldms_stream_stats_s * __stream_get_stats(struct ldms_stream_s *s)
 	LDMS_STREAM_COUNTERS_INIT(&ss->rx);
 	ss->rx = s->rx;
 
-	rc = __src_stats_rbt_copy(&s->src_stats_rbt, &ss->src_stats_rbt);
+	rc = __src_stats_rbt_copy(&s->src_stats_rbt, &ss->src_stats_rbt, is_reset);
 	if (rc)
 		goto err_1;
 
@@ -1494,7 +1496,11 @@ struct ldms_stream_stats_s * __stream_get_stats(struct ldms_stream_s *s)
 		ps->tx = sce->tx;
 		ps->drops = sce->drops;
 		TAILQ_INSERT_TAIL(&ss->pair_tq, ps, entry);
+		if (is_reset)
+			LDMS_STREAM_COUNTERS_INIT(&sce->tx);
 	}
+	if (is_reset)
+		LDMS_STREAM_COUNTERS_INIT(&s->rx);
 
 	return ss;
 
@@ -1510,7 +1516,8 @@ struct ldms_stream_stats_s * __stream_get_stats(struct ldms_stream_s *s)
 	return NULL;
 }
 
-struct ldms_stream_stats_tq_s *ldms_stream_stats_tq_get(const char *match, int is_regex)
+struct ldms_stream_stats_tq_s *
+ldms_stream_stats_tq_get(const char *match, int is_regex, int is_reset)
 {
 	regex_t r = {0};
 	int free_reg = 0;
@@ -1545,7 +1552,7 @@ struct ldms_stream_stats_tq_s *ldms_stream_stats_tq_get(const char *match, int i
 			goto done;
 		stream = container_of(rbn, struct ldms_stream_s, rbn);
 		pthread_rwlock_rdlock(&stream->rwlock);
-		stats = __stream_get_stats(stream);
+		stats = __stream_get_stats(stream, is_reset);
 		pthread_rwlock_unlock(&stream->rwlock);
 		if (!stats)
 			goto err_0;
@@ -1560,7 +1567,7 @@ struct ldms_stream_stats_tq_s *ldms_stream_stats_tq_get(const char *match, int i
 				continue;
 		}
 		pthread_rwlock_rdlock(&stream->rwlock);
-		stats = __stream_get_stats(stream);
+		stats = __stream_get_stats(stream, is_reset);
 		pthread_rwlock_unlock(&stream->rwlock);
 		if (!stats)
 			goto err_0;
@@ -1732,12 +1739,12 @@ char *ldms_stream_stats_tq_to_str(struct ldms_stream_stats_tq_s *tq)
 	return ret;
 }
 
-char *ldms_stream_stats_str(const char *match, int is_regex)
+char *ldms_stream_stats_str(const char *match, int is_regex, int is_reset)
 {
 	struct ldms_stream_stats_tq_s *tq = NULL;
 	char *ret = NULL;
 
-	tq = ldms_stream_stats_tq_get(match, is_regex);
+	tq = ldms_stream_stats_tq_get(match, is_regex, is_reset);
 	if (!tq) {
 		if (errno == ENOENT) {
 			ret = malloc(3);

ldms/src/ldmsd/ldmsd_request.c

@@ -7853,11 +7853,13 @@ static int stream_stats_handler(ldmsd_req_ctxt_t reqc)
 {
 	char *regex = ldmsd_req_attr_str_value_get_by_id(reqc, LDMSD_ATTR_REGEX);
 	char *stream = ldmsd_req_attr_str_value_get_by_id(reqc, LDMSD_ATTR_STREAM);
+	char *reset_s = ldmsd_req_attr_str_value_get_by_id(reqc, LDMSD_ATTR_RESET);
 	const char *match = NULL;
 	int is_regex = 0;
 	char buff[128];
 	char *s;
 	int rc = 0;
+	int is_reset = 0;
 	size_t len;
 	struct ldmsd_req_attr_s attr;
 
@@ -7867,7 +7869,11 @@ static int stream_stats_handler(ldmsd_req_ctxt_t reqc)
 	} else if (stream) {
 		match = stream;
 	}
-	s = ldms_stream_stats_str(match, is_regex);
+
+	if (reset_s && (0 == strcasecmp(reset_s, "true")))
+		is_reset = 1;
+
+	s = ldms_stream_stats_str(match, is_regex, is_reset);
 	if (!s) {
 		reqc->errcode = errno;
 		snprintf(buff, sizeof(buff), "ldms_stream_stats_str() error: %d",