Merge pull request #904 from Nikokolas3270/SREP-347

SREP-347 & SREP-4896: factorizing code, fixing some potential race conditions in StreamLogs & making sure 'osdctl rhobs metrics' can work at a given time or with a time range

Author: openshift-merge-bot[bot]

cmd/rhobs/cellCmd.go

@@ -39,7 +39,7 @@ func newCmdCell() *cobra.Command {
 				fmt.Println("Logs RHOBS cell   :", logsRhobsFetcher.RhobsCell)
 			}
 
-			if metricsErr != nil && logsErr != nil {
+			if metricsErr != nil || logsErr != nil {
 				os.Exit(1)
 			}
 		},

cmd/rhobs/logsCmd.go

@@ -36,12 +36,15 @@ func newCmdLogs() *cobra.Command {
 	var isPrintingTimestamp bool
 	var printedFields []string
 
-	nowTime := time.Now()
-	defaultStartTime := nowTime.Add(-5 * time.Minute)
-
 	cmd := &cobra.Command{
-		Use:           "logs [pod]",
-		Short:         "Fetch logs from RHOBS for a given cluster",
+		Use:   "logs [pod]",
+		Short: "Fetch logs from RHOBS for a given cluster",
+		Long: "Fetch logs from RHOBS for a given cluster. " +
+			"The cluster can be a management cluster (MC) or whatever cluster sending logs to RHOBS; " +
+			"the command works as if the management cluster ID was passed if given a hosted cluster (HCP) ID. " +
+			"By default, logs from all the pods in the given namespace are returned but it is possible to specify " +
+			"a single pod as an argument or filter pods using their labels. Logs themselves can be also filtered " +
+			"to only keep the ones containing a given regexp (--contain-regex option) or a given log level (--level option).",
 		Args:          cobra.RangeArgs(0, 1),
 		SilenceErrors: true,
 		RunE: func(cmd *cobra.Command, args []string) error {
@@ -152,6 +155,9 @@ func newCmdLogs() *cobra.Command {
 				}
 			}
 
+			nowTime := time.Now()
+			defaultStartTime := nowTime.Add(-5 * time.Minute)
+
 			if cmd.Flags().Changed("since") {
 				startTime = nowTime.Add(-duration)
 			} else if !cmd.Flags().Changed("start-time") && !cmd.Flags().Changed("since-time") {
@@ -206,7 +212,7 @@ func newCmdLogs() *cobra.Command {
 			}
 
 			if isFollowing {
-				err = rhobsFetcher.StreamLogs(lokiExpr, startTime, outputFormat, isPrintingTimestamp, printedFields)
+				err = rhobsFetcher.StreamLogs(lokiExpr, outputFormat, isPrintingTimestamp, printedFields)
 			} else {
 				err = rhobsFetcher.PrintLogs(lokiExpr, startTime, endTime, logsCount, isGoingForward, outputFormat, isPrintingTimestamp, printedFields)
 			}
@@ -217,7 +223,7 @@ func newCmdLogs() *cobra.Command {
 		},
 	}
 
-	cmd.Flags().StringVarP(&lokiExpr, "query", "q", "", "Loki query - exclusive with many other flags")
+	cmd.Flags().StringVarP(&lokiExpr, "query", "q", "", "LogQL expression - exclusive with many other flags")
 	cmd.Flags().StringVarP(&namespace, "namespace", "n", "default", "Name of the namespace")
 	cmd.Flags().StringVarP(&labelSelectorStr, "selector", "l", "", "Label selector for filtering pods - exclusive with the pod argument")
 	cmd.Flags().StringVarP(&containerName, "container", "c", "", "Name of the container - print all containers logs if not specified")
@@ -238,8 +244,8 @@ func newCmdLogs() *cobra.Command {
 	cmd.MarkFlagsMutuallyExclusive("query", "container")
 	cmd.MarkFlagsMutuallyExclusive("query", "include-events")
 
-	cmd.Flags().TimeVar(&startTime, "start-time", time.Time{}, []string{time.RFC3339}, "Start time for the log query - alternate alias: --since-time (default to 5 minutes ago)")
-	cmd.Flags().TimeVar(&endTime, "end-time", time.Time{}, []string{time.RFC3339}, "End time for the log query (default to now)")
+	cmd.Flags().TimeVar(&startTime, "start-time", time.Time{}, []string{time.RFC3339}, "Start time for the logs - alternate alias: --since-time (default to 5 minutes ago)")
+	cmd.Flags().TimeVar(&endTime, "end-time", time.Time{}, []string{time.RFC3339}, "End time for the logs (default to now)")
 	cmd.Flags().TimeVar(&startTime, "since-time", time.Time{}, []string{time.RFC3339}, "Same as --start-time")
 	_ = cmd.Flags().MarkHidden("since-time")
 	cmd.Flags().DurationVar(&duration, "since", 0, "Only return logs newer than a relative duration (e.g. 1h, 30m) - exclusive with --start-time & --end-time")
@@ -251,8 +257,11 @@ func newCmdLogs() *cobra.Command {
 		`"backward" returns the most recent & interesting logs first, while "forward" matches the behavior of "kubectl logs" by returning the oldest logs first `+
 		`(default to "backward" unless --follow is set in which case it is forced to "forward")`)
 
-	cmd.Flags().BoolVarP(&isFollowing, "follow", "f", false, "Specify if the logs should be streamed - exclusive with --end-time, --direction, --limit and --no-limit flags")
+	cmd.Flags().BoolVarP(&isFollowing, "follow", "f", false, "Specify if the logs should be streamed - exclusive with --start-time, --end-time, --since, --direction, --limit and --no-limit flags")
+	cmd.MarkFlagsMutuallyExclusive("follow", "start-time")
+	cmd.MarkFlagsMutuallyExclusive("follow", "since-time")
 	cmd.MarkFlagsMutuallyExclusive("follow", "end-time")
+	cmd.MarkFlagsMutuallyExclusive("follow", "since")
 	cmd.MarkFlagsMutuallyExclusive("follow", "direction")
 
 	cmd.Flags().IntVar(&logsCount, "limit", 0, "Maximum number of logs to return - allowed range: [1 100000] (default to 10000 unless --follow is set in which case there is no limit)")

cmd/rhobs/mcp_query.go

@@ -5,12 +5,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"sort"
-	"strconv"
 	"time"
 
 	rhobsclient "github.com/observatorium/api/client"
-	rhobsparameters "github.com/observatorium/api/client/parameters"
 )
 
 type mcpLogEntry struct {
@@ -19,170 +16,6 @@ type mcpLogEntry struct {
 	Stream    map[string]string `json:"stream,omitempty"`
 }
 
-func (q *RhobsFetcher) QueryInstantMetrics(ctx context.Context, promExpr string, filterCluster bool) ([]instantMetricResult, error) {
-	client, err := q.getClient()
-	if err != nil {
-		return nil, err
-	}
-
-	promQuery := rhobsparameters.PromqlQuery(promExpr)
-	queryParams := &rhobsclient.GetInstantQueryParams{Query: &promQuery}
-
-	response, err := client.GetInstantQueryWithResponse(ctx, "hcp", queryParams)
-	if err != nil {
-		return nil, fmt.Errorf("failed to send request to RHOBS: %v", err)
-	}
-	if response.HTTPResponse.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("RHOBS query failed with status code: %d - body: %s", response.HTTPResponse.StatusCode, string(response.Body))
-	}
-
-	var formattedResponse getInstantMetricsResponse
-	if err := json.Unmarshal(response.Body, &formattedResponse); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal response from RHOBS: %v", err)
-	}
-	if formattedResponse.Status != "success" {
-		return nil, fmt.Errorf("RHOBS query failed with status: %s", formattedResponse.Status)
-	}
-
-	if !filterCluster {
-		return formattedResponse.Data.Results, nil
-	}
-
-	var filtered []instantMetricResult
-	for _, result := range formattedResponse.Data.Results {
-		if q.isManagementCluster {
-			mcId := result.Metric["_mc_id"]
-			mcName := result.Metric["mc_name"]
-			if mcId == q.clusterId || mcName == q.clusterName {
-				filtered = append(filtered, result)
-			}
-		} else {
-			if result.Metric["_id"] == q.clusterExternalId {
-				filtered = append(filtered, result)
-			}
-		}
-	}
-	return filtered, nil
-}
-
-func (q *RhobsFetcher) QueryRangeMetrics(ctx context.Context, promExpr, start, end, step string) (json.RawMessage, error) {
-	client, err := q.getClient()
-	if err != nil {
-		return nil, err
-	}
-
-	promQuery := rhobsparameters.PromqlQuery(promExpr)
-	queryParams := &rhobsclient.GetRangeQueryParams{
-		Query: &promQuery,
-		Start: (*rhobsparameters.StartTS)(&start),
-		End:   (*rhobsparameters.EndTS)(&end),
-		Step:  &step,
-	}
-
-	response, err := client.GetRangeQueryWithResponse(ctx, "hcp", queryParams)
-	if err != nil {
-		return nil, fmt.Errorf("failed to send request to RHOBS: %v", err)
-	}
-	if response.HTTPResponse.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("RHOBS query failed with status code: %d - body: %s", response.HTTPResponse.StatusCode, string(response.Body))
-	}
-
-	return response.Body, nil
-}
-
-func (q *RhobsFetcher) QueryLogs(ctx context.Context, lokiExpr string, startTime, endTime time.Time, logsCount int) ([]mcpLogEntry, error) {
-	client, err := q.getClient()
-	if err != nil {
-		return nil, err
-	}
-
-	startTimeStamp := startTime.UnixNano()
-	endTimeStamp := endTime.UnixNano()
-	logsDir := "backward"
-
-	var entries []mcpLogEntry
-
-	for logsCount > 0 {
-		lokiQuery := rhobsparameters.LogqlQuery(lokiExpr)
-		startTimeStr := strconv.FormatInt(startTimeStamp, 10)
-		endTimeStr := strconv.FormatInt(endTimeStamp, 10)
-
-		limit := float32(500)
-		if logsCount < 500 {
-			limit = float32(logsCount)
-		}
-
-		queryParams := &rhobsclient.GetLogRangeQueryParams{
-			Query:     &lokiQuery,
-			Start:     (*rhobsparameters.StartTS)(&startTimeStr),
-			End:       (*rhobsparameters.EndTS)(&endTimeStr),
-			Direction: &logsDir,
-			Limit:     (*rhobsparameters.Limit)(&limit),
-		}
-
-		response, err := client.GetLogRangeQueryWithResponse(ctx, "hcp", queryParams)
-		if err != nil {
-			return nil, fmt.Errorf("failed to send request to RHOBS: %v", err)
-		}
-		if response.HTTPResponse.StatusCode != http.StatusOK {
-			return nil, fmt.Errorf("RHOBS query failed with status code: %d - body: %s", response.HTTPResponse.StatusCode, string(response.Body))
-		}
-
-		var formattedResponse getLogsResponse
-		if err := json.Unmarshal(response.Body, &formattedResponse); err != nil {
-			return nil, fmt.Errorf("failed to unmarshal response from RHOBS: %v", err)
-		}
-		if formattedResponse.Status != "success" {
-			return nil, fmt.Errorf("RHOBS query failed with status: %s", formattedResponse.Status)
-		}
-		if len(formattedResponse.Data.Results) == 0 {
-			break
-		}
-
-		var flattenedResults []*logResult
-		for _, result := range formattedResponse.Data.Results {
-			for valIdx := range result.Values {
-				flattenedResults = append(flattenedResults, &logResult{
-					Stream: result.Stream,
-					Values: []*[]string{result.Values[valIdx]},
-				})
-			}
-		}
-
-		sort.Slice(flattenedResults, func(i, j int) bool {
-			return flattenedResults[i].getTimeStamp() > flattenedResults[j].getTimeStamp()
-		})
-
-		edgeTimeStamp := flattenedResults[len(flattenedResults)-1].getTimeStamp()
-		if flattenedResults[0].getTimeStamp() == edgeTimeStamp {
-			endTimeStamp = edgeTimeStamp - 1
-			edgeTimeStamp = 0
-		} else {
-			endTimeStamp = edgeTimeStamp
-		}
-
-		for _, result := range flattenedResults {
-			ts := result.getTimeStamp()
-			if ts != edgeTimeStamp {
-				entry := mcpLogEntry{
-					Timestamp: result.getTime(),
-					Message:   result.getMessage(),
-				}
-				if result.Stream != nil {
-					entry.Stream = *result.Stream
-				}
-				entries = append(entries, entry)
-				logsCount--
-				if logsCount <= 0 {
-					break
-				}
-			}
-		}
-	}
-
-	return entries, nil
-}
-
 func (q *RhobsFetcher) QueryAlerts(ctx context.Context) (json.RawMessage, error) {
 	client, err := q.getClient()
 	if err != nil {

cmd/rhobs/mcp_tools.go

@@ -40,9 +40,8 @@ func registerMcpTools(s *mcp.Server) {
 				"cell":        {"type": "string", "description": "RHOBS cell URL (regional Thanos/Loki endpoint)"},
 				"cluster_id":  {"type": "string", "description": "Internal cluster ID"},
 				"environment": {"type": "string", "description": "OCM environment (production, stage, integration)"},
-				"results":     {"type": "array", "description": "Metric results with labels and values (instant query)"},
-				"count":       {"type": "integer", "description": "Number of results (instant query)"},
-				"result":      {"type": "object", "description": "Range query result (when start/end provided)"}
+				"results":     {"type": "array", "description": "Metric results with labels and values"},
+				"count":       {"type": "integer", "description": "Number of results"}
 			}
 		}`),
 	}, handleMetrics)
@@ -161,25 +160,24 @@ func handleMetrics(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallTool
 	}
 
 	if start != "" && end != "" {
-		rawResult, err := fetcher.QueryRangeMetrics(ctx, query, start, end, step)
+		results, err := fetcher.QueryRangeMetrics(ctx, query, NewRawMetricsTimeRange(start, end, step))
 		if err != nil {
 			return mcpError("Range metrics query failed: %v", err)
 		}
-		var parsed interface{}
-		if err := json.Unmarshal(rawResult, &parsed); err != nil {
-			cellInfo["result_raw"] = string(rawResult)
-			return mcpResultJSON(cellInfo)
-		}
-		cellInfo["result"] = parsed
+		results = filterMetricsResults(fetcher, results, filterCluster)
+		cellInfo["results"] = results
+		cellInfo["count"] = len(*results)
 		return mcpResultJSON(cellInfo)
 	}
 
-	results, err := fetcher.QueryInstantMetrics(ctx, query, filterCluster)
+	results, err := fetcher.QueryInstantMetrics(ctx, query, time.Time{})
 	if err != nil {
 		return mcpError("Metrics query failed: %v", err)
 	}
+	results = filterMetricsResults(fetcher, results, filterCluster)
+
 	cellInfo["results"] = results
-	cellInfo["count"] = len(results)
+	cellInfo["count"] = len(*results)
 	return mcpResultJSON(cellInfo)
 }
 
@@ -232,7 +230,17 @@ func handleLogs(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolRes
 	now := time.Now()
 	startTime := now.Add(-duration)
 
-	entries, err := fetcher.QueryLogs(ctx, lokiExpr, startTime, now, limit)
+	entries := []mcpLogEntry{}
+	err = fetcher.QueryLogs(ctx, lokiExpr, startTime, now, limit, false, func(result *logResult) {
+		entry := mcpLogEntry{
+			Timestamp: result.getTime(),
+			Message:   result.getMessage(),
+		}
+		if result.Stream != nil {
+			entry.Stream = *result.Stream
+		}
+		entries = append(entries, entry)
+	})
 	if err != nil {
 		return mcpError("Log query failed: %v", err)
 	}