feat: add getting cluster monitor state api

modules/elastic/api/cluster_overview.go

@@ -738,7 +738,7 @@ func (h *APIHandler) GetRealtimeClusterNodes(w http.ResponseWriter, req *http.Re
 
 func (h *APIHandler) GetClusterIndices(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	id := ps.ByName("id")
-	if GetMonitorState(id) == Console {
+	if GetMonitorState(id) == elastic.ModeAgentless {
 		h.APIHandler.GetClusterIndices(w, req, ps)
 		return
 	}
@@ -774,7 +774,7 @@ func (h *APIHandler) GetClusterIndices(w http.ResponseWriter, req *http.Request,
 func (h *APIHandler) GetRealtimeClusterIndices(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	resBody := map[string]interface{}{}
 	id := ps.ByName("id")
-	if GetMonitorState(id) == Console {
+	if GetMonitorState(id) == elastic.ModeAgentless {
 		h.APIHandler.GetRealtimeClusterIndices(w, req, ps)
 		return
 	}
@@ -1327,3 +1327,71 @@ func (h *APIHandler) SearchClusterMetadata(w http.ResponseWriter, req *http.Requ
 	}
 	w.Write(util.MustToJSONBytes(response))
 }
+
+func (h *APIHandler) getClusterMonitorState(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
+	id := ps.ByName("id")
+	collectionMode := GetMonitorState(id)
+	ret := util.MapStr{
+		"cluster_id": id,
+		"metric_collection_mode": collectionMode,
+	}
+	queryDSL := util.MapStr{
+		"query": util.MapStr{
+			"bool": util.MapStr{
+				"must": []util.MapStr{
+					{
+						"term": util.MapStr{
+							"metadata.labels.cluster_id": id,
+						},
+					},
+					{
+						"term": util.MapStr{
+							"metadata.category": "elasticsearch",
+						},
+					},
+				},
+			},
+		},
+		"size": 0,
+		"aggs": util.MapStr{
+			"grp_name": util.MapStr{
+				"terms": util.MapStr{
+					"field": "metadata.name",
+					"size": 10,
+				},
+				"aggs": util.MapStr{
+					"max_timestamp": util.MapStr{
+						"max": util.MapStr{
+							"field": "timestamp",
+						},
+					},
+				},
+			},
+		},
+	}
+	dsl := util.MustToJSONBytes(queryDSL)
+	response, err := elastic.GetClient(global.MustLookupString(elastic.GlobalSystemElasticsearchID)).SearchWithRawQueryDSL(getAllMetricsIndex(), dsl)
+	if err != nil {
+		log.Error(err)
+		h.WriteError(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	lastActiveAt := util.MapStr{}
+	for _, bk := range response.Aggregations["grp_name"].Buckets {
+		key := bk["key"].(string)
+		if tv, ok := bk["max_timestamp"].(map[string]interface{}); ok {
+			if collectionMode == elastic.ModeAgentless {
+				if util.StringInArray([]string{ "index_stats", "cluster_health", "cluster_stats", "node_stats"}, key) {
+					lastActiveAt[key] = tv["value"]
+				}
+			}else{
+				if util.StringInArray([]string{ "shard_stats", "cluster_health", "cluster_stats", "node_stats"}, key) {
+					lastActiveAt[key] = tv["value"]
+				}
+			}
+		}
+
+	}
+	ret["last_active_at"] = lastActiveAt
+	h.WriteJSON(w, ret, http.StatusOK)
+}

modules/elastic/api/index_overview.go

@@ -303,7 +303,7 @@ func (h *APIHandler) FetchIndexInfo(w http.ResponseWriter,  req *http.Request, p
 			return
 		}
 		firstClusterID, firstIndexName = parts[0], parts[1]
-		if GetMonitorState(firstClusterID) == Console {
+		if GetMonitorState(firstClusterID) == elastic.ModeAgentless {
 			h.APIHandler.FetchIndexInfo(w, ctx, indexIDs)
 			return
 		}
@@ -580,7 +580,7 @@ func (h *APIHandler) FetchIndexInfo(w http.ResponseWriter,  req *http.Request, p
 
 func (h *APIHandler) GetIndexInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	clusterID := ps.MustGetParameter("id")
-	if GetMonitorState(clusterID) == Console {
+	if GetMonitorState(clusterID) == elastic.ModeAgentless {
 		h.APIHandler.GetIndexInfo(w, req, ps)
 		return
 	}
@@ -701,7 +701,7 @@ func (h *APIHandler) GetIndexInfo(w http.ResponseWriter, req *http.Request, ps h
 
 func (h *APIHandler) GetIndexShards(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	clusterID := ps.MustGetParameter("id")
-	if GetMonitorState(clusterID) == Console {
+	if GetMonitorState(clusterID) == elastic.ModeAgentless {
 		h.APIHandler.GetIndexShards(w, req, ps)
 		return
 	}
@@ -810,7 +810,7 @@ func (h *APIHandler) GetIndexShards(w http.ResponseWriter, req *http.Request, ps
 
 func (h *APIHandler) GetSingleIndexMetrics(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	clusterID := ps.MustGetParameter("id")
-	if GetMonitorState(clusterID) == Console {
+	if GetMonitorState(clusterID) == elastic.ModeAgentless {
 		h.APIHandler.GetSingleIndexMetrics(w, req, ps)
 		return
 	}

modules/elastic/api/init.go

@@ -86,6 +86,7 @@ func init() {
 	api.HandleAPIMethod(api.POST, "/elasticsearch/cluster/info", clusterAPI.RequirePermission(clusterAPI.FetchClusterInfo, enum.PermissionElasticsearchMetricRead))
 
 	api.HandleAPIMethod(api.GET, "/elasticsearch/:id/info", clusterAPI.RequireClusterPermission(clusterAPI.RequirePermission(clusterAPI.GetClusterInfo, enum.PermissionElasticsearchMetricRead)))
+	api.HandleAPIMethod(api.GET, "/elasticsearch/:id/monitor_state", clusterAPI.RequireClusterPermission(clusterAPI.RequirePermission(clusterAPI.getClusterMonitorState, enum.PermissionElasticsearchMetricRead)))
 	api.HandleAPIMethod(api.POST, "/elasticsearch/node/_search", clusterAPI.RequirePermission(clusterAPI.SearchNodeMetadata, enum.PermissionElasticsearchNodeRead))
 	api.HandleAPIMethod(api.GET, "/elasticsearch/:id/nodes", clusterAPI.RequireClusterPermission(clusterAPI.RequirePermission(clusterAPI.GetClusterNodes, enum.PermissionElasticsearchMetricRead, enum.PermissionElasticsearchNodeRead)))
 	api.HandleAPIMethod(api.GET, "/elasticsearch/:id/nodes/realtime", clusterAPI.RequireClusterPermission(clusterAPI.RequirePermission(clusterAPI.GetRealtimeClusterNodes, enum.PermissionElasticsearchMetricRead)))

modules/elastic/api/manage.go

@@ -528,7 +528,7 @@ func (h *APIHandler) HandleMetricsSummaryAction(w http.ResponseWriter, req *http
 func (h *APIHandler) HandleClusterMetricsAction(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	resBody := map[string]interface{}{}
 	id := ps.ByName("id")
-	if GetMonitorState(id) == Console {
+	if GetMonitorState(id) == elastic.ModeAgentless {
 		h.APIHandler.HandleClusterMetricsAction(w, req, ps)
 		return
 	}
@@ -626,7 +626,7 @@ func (h *APIHandler) HandleNodeMetricsAction(w http.ResponseWriter, req *http.Re
 func (h *APIHandler) HandleIndexMetricsAction(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	resBody := map[string]interface{}{}
 	id := ps.ByName("id")
-	if GetMonitorState(id) == Console {
+	if GetMonitorState(id) == elastic.ModeAgentless {
 		h.APIHandler.HandleIndexMetricsAction(w, req, ps)
 		return
 	}

modules/elastic/api/monitor_state.go

@@ -32,18 +32,13 @@ import (
 	"infini.sh/framework/core/elastic"
 )
 
-type MonitorState int
-const (
-	Console MonitorState = iota
-	Agent
-)
-func GetMonitorState(clusterID string) MonitorState {
+func GetMonitorState(clusterID string) string {
 	conf := elastic.GetConfig(clusterID)
 	if conf == nil {
 		panic(fmt.Errorf("config of cluster [%s] is not found", clusterID))
 	}
 	if conf.MonitorConfigs != nil && !conf.MonitorConfigs.NodeStats.Enabled && !conf.MonitorConfigs.IndexStats.Enabled {
-		return Agent
+		return elastic.ModeAgent
 	}
-	return Console
+	return elastic.ModeAgentless
 }

modules/elastic/api/node_overview.go

@@ -1276,7 +1276,7 @@ func (h *APIHandler) getLatestIndices(req *http.Request, min string, max string,
 
 func (h *APIHandler) GetNodeShards(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
 	clusterID := ps.MustGetParameter("id")
-	if GetMonitorState(clusterID) == Console {
+	if GetMonitorState(clusterID) == elastic.ModeAgentless {
 		h.APIHandler.GetNodeShards(w, req, ps)
 		return
 	}