Merge pull request #1168 from milroy/match-stats

Update match stats and implement stats-clear

Author: mergify[bot]

resource/modules/resource_match.cpp

@@ -44,9 +44,15 @@ using namespace Flux::opts_manager;
 struct match_perf_t {
     double load;                   /* Graph load time */
     uint64_t njobs;                /* Total match count */
+    uint64_t njobs_reset;          /* Jobs since reset match count */
+    /* Graph uptime in seconds */
+    std::chrono::time_point<std::chrono::system_clock> graph_uptime;
+    /* Time since stats were last cleared */
+    std::chrono::time_point<std::chrono::system_clock> time_since_reset;
     double min;                    /* Min match time */
     double max;                    /* Max match time */
-    double accum;                  /* Total match time accumulated */
+    double mean;                   /* Mean match time */
+    double M2;                     /* Welford's algorithm */
 };
 
 class msg_wrap_t {
@@ -234,6 +240,9 @@ static void info_request_cb (flux_t *h, flux_msg_handler_t *w,
 static void stat_request_cb (flux_t *h, flux_msg_handler_t *w,
                              const flux_msg_t *msg, void *arg);
 
+static void stat_clear_cb (flux_t *h, flux_msg_handler_t *w,
+                             const flux_msg_t *msg, void *arg);
+
 static void next_jobid_request_cb (flux_t *h, flux_msg_handler_t *w,
                                    const flux_msg_t *msg, void *arg);
 
@@ -279,7 +288,9 @@ static const struct flux_msg_handler_spec htab[] = {
     { FLUX_MSGTYPE_REQUEST,
       "sched-fluxion-resource.info", info_request_cb, 0 },
     { FLUX_MSGTYPE_REQUEST,
-      "sched-fluxion-resource.stat", stat_request_cb, 0 },
+      "sched-fluxion-resource.stats-get", stat_request_cb, FLUX_ROLE_USER },
+    { FLUX_MSGTYPE_REQUEST,
+      "sched-fluxion-resource.stats-clear", stat_clear_cb, FLUX_ROLE_USER },
     { FLUX_MSGTYPE_REQUEST,
       "sched-fluxion-resource.next_jobid", next_jobid_request_cb, 0 },
     { FLUX_MSGTYPE_REQUEST,
@@ -344,9 +355,13 @@ static std::shared_ptr<resource_ctx_t> getctx (flux_t *h)
         set_default_args (ctx);
         ctx->perf.load = 0.0f;
         ctx->perf.njobs = 0;
-        ctx->perf.min = std::numeric_limits<double>::max();
+        ctx->perf.njobs_reset = 0;
+        ctx->perf.min = std::numeric_limits<double>::max ();
         ctx->perf.max = 0.0f;
-        ctx->perf.accum = 0.0f;
+        ctx->perf.mean = 0.0f;
+        ctx->perf.M2 = 0.0f;
+        ctx->perf.graph_uptime = std::chrono::system_clock::now ();
+        ctx->perf.time_since_reset = std::chrono::system_clock::now ();
         ctx->matcher = nullptr; /* Cannot be allocated at this point */
         ctx->fgraph = nullptr;  /* Cannot be allocated at this point */
         ctx->writers = nullptr; /* Cannot be allocated at this point */
@@ -356,7 +371,6 @@ static std::shared_ptr<resource_ctx_t> getctx (flux_t *h)
         ctx->m_r_alloc = nullptr;
         ctx->m_resources_updated = true;
         ctx->m_resources_down_updated = true;
-        //gettimeofday (&(ctx->m_resources_alloc_updated), NULL);
         ctx->m_resources_alloc_updated = std::chrono::system_clock::now ();
     }
 
@@ -1351,6 +1365,7 @@ static int populate_resource_db (std::shared_ptr<resource_ctx_t> &ctx)
 
     elapsed = std::chrono::system_clock::now () - start;
     ctx->perf.load = elapsed.count ();
+    ctx->perf.graph_uptime = std::chrono::system_clock::now ();
     rc = 0;
 
 done:
@@ -1505,10 +1520,17 @@ static int init_resource_graph (std::shared_ptr<resource_ctx_t> &ctx)
 static void update_match_perf (std::shared_ptr<resource_ctx_t> &ctx,
                                double elapse)
 {
+    double delta = 0.0f;
+    double delta2 = 0.2f;
     ctx->perf.njobs++;
+    ctx->perf.njobs_reset++;
     ctx->perf.min = (ctx->perf.min > elapse)? elapse : ctx->perf.min;
     ctx->perf.max = (ctx->perf.max < elapse)? elapse : ctx->perf.max;
-    ctx->perf.accum += elapse;
+    // Welford's online algorithm for variance
+    delta = elapse - ctx->perf.mean;
+    ctx->perf.mean += delta / (double)ctx->perf.njobs;
+    delta2 = elapse - ctx->perf.mean;
+    ctx->perf.M2 += delta * delta2;
 }
 
 static inline std::string get_status_string (int64_t now, int64_t at)
@@ -2141,12 +2163,18 @@ static void stat_request_cb (flux_t *h, flux_msg_handler_t *w,
     std::shared_ptr<resource_ctx_t> ctx = getctx ((flux_t *)arg);
     int saved_errno;
     json_t *o = nullptr;
-    double avg = 0.0f;
+    double mean = 0.0f;
     double min = 0.0f;
+    double variance = 0.0f;
+    int64_t graph_uptime_s = 0;
+    int64_t time_since_reset_s = 0;
+    std::chrono::time_point<std::chrono::system_clock> now;
 
-    if (ctx->perf.njobs) {
-        avg = ctx->perf.accum / (double)ctx->perf.njobs;
+    if (ctx->perf.njobs_reset > 1) {
+        mean = ctx->perf.mean;
         min = ctx->perf.min;
+        // Welford's online algorithm
+        variance = ctx->perf.M2 / (double)ctx->perf.njobs_reset;
     }
     if ( !(o = json_object ())) {
         errno = ENOMEM;
@@ -2156,15 +2184,24 @@ static void stat_request_cb (flux_t *h, flux_msg_handler_t *w,
         flux_log_error (h, "%s: get_stat_by_rank", __FUNCTION__);
         goto error_free;
     }
-    if (flux_respond_pack (h, msg, "{s:I s:I s:o s:f s:I s:f s:f s:f}",
+    now = std::chrono::system_clock::now ();
+    graph_uptime_s = std::chrono::duration_cast<std::chrono::seconds> (
+                        now - ctx->perf.graph_uptime).count ();
+    time_since_reset_s = std::chrono::duration_cast<std::chrono::seconds> (
+                        now - ctx->perf.time_since_reset).count ();
+    if (flux_respond_pack (h, msg, "{s:I s:I s:o s:f s:I s:I s:I s:I s:f s:f s:f s:f}",
                                    "V", num_vertices (ctx->db->resource_graph),
                                    "E", num_edges (ctx->db->resource_graph),
                                    "by_rank", o,
                                    "load-time", ctx->perf.load,
+                                   "graph-uptime", graph_uptime_s,
+                                   "time-since-reset", time_since_reset_s,
                                    "njobs", ctx->perf.njobs,
+                                   "njobs-reset", ctx->perf.njobs_reset,
                                    "min-match", min,
                                    "max-match", ctx->perf.max,
-                                   "avg-match", avg) < 0) {
+                                   "avg-match", mean,
+                                   "match-variance", variance) < 0) {
         flux_log_error (h, "%s: flux_respond_pack", __FUNCTION__);
     }
 
@@ -2179,6 +2216,23 @@ static void stat_request_cb (flux_t *h, flux_msg_handler_t *w,
         flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
 }
 
+static void stat_clear_cb (flux_t *h, flux_msg_handler_t *w,
+                           const flux_msg_t *msg, void *arg)
+{
+    std::shared_ptr<resource_ctx_t> ctx = getctx ((flux_t *)arg);
+
+    // Clear the jobs-related stats and reset time
+    ctx->perf.njobs_reset = 0;
+    ctx->perf.min = std::numeric_limits<double>::max ();
+    ctx->perf.max = 0.0f;
+    ctx->perf.mean = 0.0f;
+    ctx->perf.M2 = 0.0f;
+    ctx->perf.time_since_reset = std::chrono::system_clock::now ();
+
+    if (flux_respond (h, msg, NULL) < 0)
+        flux_log_error (h, "%s: flux_respond", __FUNCTION__);
+}
+
 static inline int64_t next_jobid (const std::map<uint64_t,
                                             std::shared_ptr<job_info_t>> &m)
 {

resource/reapi/bindings/c++/reapi_module_impl.hpp

@@ -280,7 +280,7 @@ int reapi_module_t::stat (void *h, int64_t &V, int64_t &E,int64_t &J,
         goto out;
     }
 
-    if (!(f = flux_rpc (fh, "sched-fluxion-resource.stat",
+    if (!(f = flux_rpc (fh, "sched-fluxion-resource.stats-get",
                         NULL, FLUX_NODEID_ANY, 0))) {
         goto out;
     }

src/cmd/flux-ion-resource.py

@@ -69,7 +69,10 @@ def rpc_info(self, jobid):
         return self.handle.rpc("sched-fluxion-resource.info", payload).get()
 
     def rpc_stat(self):
-        return self.handle.rpc("sched-fluxion-resource.stat").get()
+        return self.handle.rpc("sched-fluxion-resource.stats-get").get()
+
+    def rpc_stats_clear(self):
+        return self.handle.rpc("sched-fluxion-resource.stats-clear").get()
 
     def rpc_cancel(self, jobid):
         payload = {"jobid": jobid}
@@ -219,10 +222,23 @@ def stat_action(_):
     print("Num. of Edges: ", resp["E"])
     print("Num. of Vertices by Rank: ", json.dumps(resp["by_rank"]))
     print("Graph Load Time: ", resp["load-time"], "Secs")
-    print("Num. of Jobs Matched: ", resp["njobs"])
+    print("Graph Upime: ", resp["graph-uptime"], "Secs")
+    print("Time Since Stats Reset: ", resp["time-since-reset"], "Secs")
+    print("Num. of Total Jobs Matched: ", resp["njobs"])
+    print("Num. of Jobs Matched Since Reset: ", resp["njobs-reset"])
     print("Min. Match Time: ", resp["min-match"], "Secs")
     print("Max. Match Time: ", resp["max-match"], "Secs")
     print("Avg. Match Time: ", resp["avg-match"], "Secs")
+    print("Match Variance: ", resp["match-variance"], "Secs^2")
+
+
+def stats_clear_action(_):
+    """
+    Action for stats clear sub-command
+    """
+
+    rmod = ResourceModuleInterface()
+    rmod.rpc_stats_clear()
 
 
 def set_property_action(args):
@@ -417,6 +433,7 @@ def parse_set_status(parser_ss: argparse.ArgumentParser):
     parser_ss.set_defaults(func=set_status_action)
 
 
+# pylint: disable=too-many-statements
 def main():
     """
     Main entry point
@@ -447,7 +464,8 @@ def mkparser(name, help_desc):
     parse_match(mkparser("match", "Find the best matching resources for a jobspec."))
     parse_update(mkparser("update", "Update the resource database."))
     parse_info(mkparser("info", "Print info on a single job."))
-    parser_s = mkparser("stat", "Print overall performance statistics.")
+    parser_s = mkparser("stats", "Print overall performance statistics.")
+    parser_sc = mkparser("stats-cancel", "Clear overall performance statistics.")
     parser_c = mkparser("cancel", "Cancel an allocated or reserved job.")
     parse_find(mkparser("find", "Find resources matching with a criteria."))
     parser_st = mkparser("status", "Display resource status.")
@@ -466,6 +484,11 @@ def mkparser(name, help_desc):
     #
     parser_s.set_defaults(func=stat_action)
 
+    #
+    # Action for stats-clear sub-command
+    #
+    parser_sc.set_defaults(func=stats_clear_action)
+
     #
     # Positional argument for cancel sub-command
     #

t/t4000-match-params.t

@@ -100,6 +100,14 @@ test_expect_success 'loading resource module with unknown policies is tolerated'
     load_resource policy=bar
 '
 
+test_expect_success 'resource module stats and clear work' '
+    unload_resource &&
+    load_resource &&
+    load_qmanager_sync &&
+    flux module stats sched-fluxion-resource &&
+    flux module stats --clear sched-fluxion-resource
+'
+
 test_expect_success 'removing resource works' '
     remove_resource
 '

t/t6002-graph-hwloc.t

@@ -30,7 +30,7 @@ test_expect_success 'qmanager: loading resource and qmanager modules works' '
 '
 
 test_expect_success 'qmanager: graph stat as expected' '
-    flux ion-resource stat > stat.out &&
+    flux ion-resource stats > stat.out &&
     test_debug "cat stat.out" &&
     verify stat.out
 '