Indexes being written to are sorted to the top now

elastop.go

@@ -207,10 +207,11 @@ type CatNodesStats struct {
 }
 
 var (
-	showNodes   = true
-	showRoles   = true
-	showIndices = true
-	showMetrics = true
+	showNodes         = true
+	showRoles         = true
+	showIndices       = true
+	showMetrics       = true
+	showHiddenIndices = false
 )
 
 var (
@@ -221,6 +222,21 @@ var (
 	metricsPanel *tview.TextView
 )
 
+type DataStreamResponse struct {
+	DataStreams []DataStream `json:"data_streams"`
+}
+
+type DataStream struct {
+	Name      string `json:"name"`
+	Timestamp string `json:"timestamp"`
+	Status    string `json:"status"`
+	Template  string `json:"template"`
+}
+
+var (
+	apiKey string
+)
+
 func bytesToHuman(bytes int64) string {
 	const unit = 1024
 	if bytes < unit {
@@ -455,28 +471,15 @@ func updateGridLayout(grid *tview.Grid, showRoles, showIndices, showMetrics bool
 	// Start with clean grid
 	grid.Clear()
 
-	visiblePanels := make([]struct {
-		panel *tview.TextView
-		show  bool
-	}, 0)
-
+	visiblePanels := 0
 	if showRoles {
-		visiblePanels = append(visiblePanels, struct {
-			panel *tview.TextView
-			show  bool
-		}{rolesPanel, true})
+		visiblePanels++
 	}
 	if showIndices {
-		visiblePanels = append(visiblePanels, struct {
-			panel *tview.TextView
-			show  bool
-		}{indicesPanel, true})
+		visiblePanels++
 	}
 	if showMetrics {
-		visiblePanels = append(visiblePanels, struct {
-			panel *tview.TextView
-			show  bool
-		}{metricsPanel, true})
+		visiblePanels++
 	}
 
 	// Adjust row configuration based on whether nodes panel is shown
@@ -486,40 +489,63 @@ func updateGridLayout(grid *tview.Grid, showRoles, showIndices, showMetrics bool
 		grid.SetRows(3, 0) // Just header and bottom panels
 	}
 
-	// Set up columns based on number of visible panels
-	switch len(visiblePanels) {
-	case 3:
-		grid.SetColumns(-1, -2, -1) // 1:2:1 ratio
-	case 2:
-		grid.SetColumns(-1, -1) // Two equal columns
-	case 1:
-		grid.SetColumns(-1) // Single column
+	// Configure columns based on visible panels
+	switch {
+	case visiblePanels == 3:
+		if showRoles {
+			grid.SetColumns(30, -2, -1) // Changed from 20 to 30 for roles panel width
+		}
+	case visiblePanels == 2:
+		if showRoles {
+			grid.SetColumns(30, 0) // Changed from 20 to 30 for roles panel width
+		} else {
+			grid.SetColumns(-1, -1) // Equal split between two panels
+		}
+	case visiblePanels == 1:
+		grid.SetColumns(0) // Single column takes full width
 	}
 
 	// Always show header at top spanning all columns
-	grid.AddItem(header, 0, 0, 1, len(visiblePanels), 0, 0, false)
+	grid.AddItem(header, 0, 0, 1, visiblePanels, 0, 0, false)
 
 	// Add nodes panel if visible, spanning all columns
 	if showNodes {
-		grid.AddItem(nodesPanel, 1, 0, 1, len(visiblePanels), 0, 0, false)
+		grid.AddItem(nodesPanel, 1, 0, 1, visiblePanels, 0, 0, false)
+	}
 
-		// Add bottom panels
-		for i, panel := range visiblePanels {
-			grid.AddItem(panel.panel, 2, i, 1, 1, 0, 0, false)
+	// Add bottom panels in their respective positions
+	col := 0
+	if showRoles {
+		row := 1
+		if showNodes {
+			row = 2
 		}
-	} else {
-		// Add bottom panels starting from row 1
-		for i, panel := range visiblePanels {
-			grid.AddItem(panel.panel, 1, i, 1, 1, 0, 0, false)
+		grid.AddItem(rolesPanel, row, col, 1, 1, 0, 0, false)
+		col++
+	}
+	if showIndices {
+		row := 1
+		if showNodes {
+			row = 2
 		}
+		grid.AddItem(indicesPanel, row, col, 1, 1, 0, 0, false)
+		col++
+	}
+	if showMetrics {
+		row := 1
+		if showNodes {
+			row = 2
+		}
+		grid.AddItem(metricsPanel, row, col, 1, 1, 0, 0, false)
 	}
 }
 
 func main() {
 	host := flag.String("host", "http://localhost", "Elasticsearch host URL (e.g., http://localhost or https://example.com)")
 	port := flag.Int("port", 9200, "Elasticsearch port")
-	user := flag.String("user", "elastic", "Elasticsearch username")
+	user := flag.String("user", os.Getenv("ES_USER"), "Elasticsearch username")
 	password := flag.String("password", os.Getenv("ES_PASSWORD"), "Elasticsearch password")
+	flag.StringVar(&apiKey, "apikey", os.Getenv("ES_API_KEY"), "Elasticsearch API key")
 	flag.Parse()
 
 	// Validate and process the host URL
@@ -528,6 +554,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Validate authentication
+	if apiKey != "" && (*user != "" || *password != "") {
+		fmt.Fprintf(os.Stderr, "Error: Cannot use both API key and username/password authentication\n")
+		os.Exit(1)
+	}
+
+	if apiKey == "" && (*user == "" || *password == "") {
+		fmt.Fprintf(os.Stderr, "Error: Must provide either API key or both username and password\n")
+		os.Exit(1)
+	}
+
 	// Strip any trailing slash from the host
 	*host = strings.TrimRight(*host, "/")
 
@@ -544,7 +581,7 @@ func main() {
 
 	app := tview.NewApplication()
 
-	// Update the grid layout to use three columns for the bottom section
+	// Update the grid layout to use proportional columns
 	grid := tview.NewGrid().
 		SetRows(3, 0, 0).       // Three rows: header, nodes, bottom panels
 		SetColumns(-1, -2, -1). // Three columns for bottom row: roles (1), indices (2), metrics (1)
@@ -587,12 +624,25 @@ func main() {
 			if err != nil {
 				return err
 			}
-			req.SetBasicAuth(*user, *password)
+
+			// Set authentication
+			if apiKey != "" {
+				req.Header.Set("Authorization", fmt.Sprintf("ApiKey %s", apiKey))
+			} else {
+				req.SetBasicAuth(*user, *password)
+			}
+
 			resp, err := client.Do(req)
 			if err != nil {
 				return err
 			}
 			defer resp.Body.Close()
+
+			if resp.StatusCode != http.StatusOK {
+				body, _ := io.ReadAll(resp.Body)
+				return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body))
+			}
+
 			body, err := io.ReadAll(resp.Body)
 			if err != nil {
 				return err
@@ -681,7 +731,7 @@ func main() {
 		}[clusterStats.Status]
 
 		// Get max lengths after fetching node and index info
-		maxNodeNameLen, maxIndexNameLen := getMaxLengths(nodesInfo, indicesStats)
+		maxNodeNameLen, maxIndexNameLen, maxTransportLen := getMaxLengths(nodesInfo, indicesStats)
 
 		// Update header with dynamic padding
 		header.Clear()
@@ -690,25 +740,35 @@ func main() {
 		if maxNodeNameLen > len(clusterStats.ClusterName) {
 			padding = maxNodeNameLen - len(clusterStats.ClusterName)
 		}
-		fmt.Fprintf(header, "[#00ffff]Cluster:[white] %s [%s]%s[-]%s[#00ffff]Latest: [white]%s\n",
+		fmt.Fprintf(header, "[#00ffff]Cluster :[white] %s [#666666]([%s]%s[-]%s[#666666]) [#00ffff]Latest: [white]%s\n",
 			clusterStats.ClusterName,
 			statusColor,
 			strings.ToUpper(clusterStats.Status),
 			strings.Repeat(" ", padding),
 			latestVer)
-		fmt.Fprintf(header, "[#00ffff]Nodes:[white] %d Total, [green]%d[white] Successful, [#ff5555]%d[white] Failed\n",
+		fmt.Fprintf(header, "[#00ffff]Nodes   :[white] %d Total, [green]%d[white] Successful, [#ff5555]%d[white] Failed\n",
 			clusterStats.Nodes.Total,
 			clusterStats.Nodes.Successful,
 			clusterStats.Nodes.Failed)
-		fmt.Fprintf(header, "[#666666]Press 2-5 to toggle panels, 'q' to quit[white]\n")
+		fmt.Fprintf(header, "[#666666]Press 2-5 to toggle panels, 'h' to toggle hidden indices, 'q' to quit[white]\n")
 
 		// Update nodes panel with dynamic width
 		nodesPanel.Clear()
 		fmt.Fprintf(nodesPanel, "[::b][#00ffff][[#ff5555]2[#00ffff]] Nodes Information[::-]\n\n")
-		fmt.Fprint(nodesPanel, getNodesPanelHeader(maxNodeNameLen))
+		fmt.Fprint(nodesPanel, getNodesPanelHeader(maxNodeNameLen, maxTransportLen))
+
+		// Create a sorted slice of node IDs based on node names
+		var nodeIDs []string
+		for id := range nodesInfo.Nodes {
+			nodeIDs = append(nodeIDs, id)
+		}
+		sort.Slice(nodeIDs, func(i, j int) bool {
+			return nodesInfo.Nodes[nodeIDs[i]].Name < nodesInfo.Nodes[nodeIDs[j]].Name
+		})
 
 		// Update node entries with dynamic width
-		for id, nodeInfo := range nodesInfo.Nodes {
+		for _, id := range nodeIDs {
+			nodeInfo := nodesInfo.Nodes[id]
 			nodeStats, exists := nodesStats.Nodes[id]
 			if !exists {
 				continue
@@ -752,10 +812,29 @@ func main() {
 				nodeLoads[node.Name] = node.Load1m
 			}
 
-			fmt.Fprintf(nodesPanel, "[#5555ff]%-*s  [white] [#444444]│[white] %s [#444444]│[white] [white]%-20s[white] [#444444]│[white] [%s]%-7s[white] [#444444]│[white] [%s]%3d%% [#444444](%d)[white] [#444444]│[white] %4s [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %s [#bd93f9]%s[white] [#444444](%s)[white]\n",
+			// In the update() function, add this request before processing nodes:
+			var threadPoolStats []ThreadPoolStats
+			if err := makeRequest("/_cat/thread_pool/generic?format=json&h=node_name,name,active,queue,rejected,completed", &threadPoolStats); err != nil {
+				nodesPanel.SetText(fmt.Sprintf("[red]Error getting thread pool stats: %v", err))
+				return
+			}
+
+			// Create a map for quick lookup of thread pool stats by node name
+			threadPoolMap := make(map[string]ThreadPoolStats)
+			for _, stat := range threadPoolStats {
+				threadPoolMap[stat.NodeName] = stat
+			}
+
+			active, _ := strconv.Atoi(threadPoolMap[nodeInfo.Name].Active)
+			queue, _ := strconv.Atoi(threadPoolMap[nodeInfo.Name].Queue)
+			rejected, _ := strconv.Atoi(threadPoolMap[nodeInfo.Name].Rejected)
+			completed, _ := strconv.Atoi(threadPoolMap[nodeInfo.Name].Completed)
+
+			fmt.Fprintf(nodesPanel, "[#5555ff]%-*s  [white] [#444444]│[white] %s [#444444]│[white] [white]%-*s[white] [#444444]│[white] [%s]%-7s[white] [#444444]│[white] [%s]%3d%% [#444444](%d)[white] [#444444]│[white] %4s [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %4s / %4s [%s]%3d%%[white] [#444444]│[white] %6s [#444444]│[white] %5s [#444444]│[white] %8s [#444444]│[white] %9s [#444444]│[white] %s [#bd93f9]%s[white] [#444444](%s)[white]\n",
 				maxNodeNameLen,
 				nodeInfo.Name,
 				formatNodeRoles(nodeInfo.Roles),
+				maxTransportLen,
 				nodeInfo.TransportAddress,
 				versionColor,
 				nodeInfo.Version,
@@ -775,11 +854,22 @@ func main() {
 				formatResourceSize(diskTotal),
 				getPercentageColor(diskPercent),
 				int(diskPercent),
+				formatNumber(active),
+				formatNumber(queue),
+				formatNumber(rejected),
+				formatNumber(completed),
 				nodeInfo.OS.PrettyName,
 				nodeInfo.OS.Version,
 				nodeInfo.OS.Arch)
 		}
 
+		// Get data streams info
+		var dataStreamResp DataStreamResponse
+		if err := makeRequest("/_data_stream", &dataStreamResp); err != nil {
+			indicesPanel.SetText(fmt.Sprintf("[red]Error getting data streams: %v", err))
+			return
+		}
+
 		// Update indices panel with dynamic width
 		indicesPanel.Clear()
 		fmt.Fprintf(indicesPanel, "[::b][#00ffff][[#ff5555]4[#00ffff]] Indices Information[::-]\n\n")
@@ -792,48 +882,49 @@ func main() {
 
 		// Collect index information
 		for _, index := range indicesStats {
-			// Skip hidden indices
-			if !strings.HasPrefix(index.Index, ".") && index.DocsCount != "0" {
-				docs := 0
-				fmt.Sscanf(index.DocsCount, "%d", &docs)
-				totalDocs += docs
-
-				// Track document changes
-				activity, exists := indexActivities[index.Index]
-				if !exists {
-					indexActivities[index.Index] = &IndexActivity{
-						LastDocsCount:    docs,
-						InitialDocsCount: docs,
-						StartTime:        time.Now(),
-					}
-				} else {
-					activity.LastDocsCount = docs
-				}
-
-				// Get write operations count and calculate rate
-				writeOps := int64(0)
-				indexingRate := float64(0)
-				if stats, exists := indexWriteStats.Indices[index.Index]; exists {
-					writeOps = stats.Total.Indexing.IndexTotal
-					if activity, ok := indexActivities[index.Index]; ok {
-						timeDiff := time.Since(activity.StartTime).Seconds()
-						if timeDiff > 0 {
-							indexingRate = float64(docs-activity.InitialDocsCount) / timeDiff
-						}
-					}
-				}
-
-				indices = append(indices, indexInfo{
-					index:        index.Index,
-					health:       index.Health,
-					docs:         docs,
-					storeSize:    index.StoreSize,
-					priShards:    index.PriShards,
-					replicas:     index.Replicas,
-					writeOps:     writeOps,
-					indexingRate: indexingRate,
-				})
+			// Skip hidden indices unless showHiddenIndices is true
+			if (!showHiddenIndices && strings.HasPrefix(index.Index, ".")) || index.DocsCount == "0" {
+				continue
 			}
+			docs := 0
+			fmt.Sscanf(index.DocsCount, "%d", &docs)
+			totalDocs += docs
+
+			// Track document changes
+			activity, exists := indexActivities[index.Index]
+			if !exists {
+				indexActivities[index.Index] = &IndexActivity{
+					LastDocsCount:    docs,
+					InitialDocsCount: docs,
+					StartTime:        time.Now(),
+				}
+			} else {
+				activity.LastDocsCount = docs
+			}
+
+			// Get write operations count and calculate rate
+			writeOps := int64(0)
+			indexingRate := float64(0)
+			if stats, exists := indexWriteStats.Indices[index.Index]; exists {
+				writeOps = stats.Total.Indexing.IndexTotal
+				if activity, ok := indexActivities[index.Index]; ok {
+					timeDiff := time.Since(activity.StartTime).Seconds()
+					if timeDiff > 0 {
+						indexingRate = float64(docs-activity.InitialDocsCount) / timeDiff
+					}
+				}
+			}
+
+			indices = append(indices, indexInfo{
+				index:        index.Index,
+				health:       index.Health,
+				docs:         docs,
+				storeSize:    index.StoreSize,
+				priShards:    index.PriShards,
+				replicas:     index.Replicas,
+				writeOps:     writeOps,
+				indexingRate: indexingRate,
+			})
 		}
 
 		// Calculate total size
@@ -841,27 +932,37 @@ func main() {
 			totalSize += node.FS.Total.TotalInBytes - node.FS.Total.AvailableInBytes
 		}
 
-		// Sort indices by document count (descending)
+		// Sort indices - active ones first, then alphabetically within each group
 		sort.Slice(indices, func(i, j int) bool {
-			return indices[i].docs > indices[j].docs
+			// If one is active and the other isn't, active goes first
+			if (indices[i].indexingRate > 0) != (indices[j].indexingRate > 0) {
+				return indices[i].indexingRate > 0
+			}
+			// Within the same group (both active or both inactive), sort alphabetically
+			return indices[i].index < indices[j].index
 		})
 
 		// Update index entries with dynamic width
 		for _, idx := range indices {
-			// Only show purple dot if there's actual indexing happening
 			writeIcon := "[#444444]⚪"
 			if idx.indexingRate > 0 {
 				writeIcon = "[#5555ff]⚫"
 			}
 
+			// Add data stream indicator
+			streamIndicator := " "
+			if isDataStream(idx.index, dataStreamResp) {
+				streamIndicator = "[#bd93f9]⚡[white]"
+			}
+
 			// Calculate document changes
 			activity := indexActivities[idx.index]
 			ingestedStr := ""
 			if activity != nil && activity.InitialDocsCount < idx.docs {
 				docChange := idx.docs - activity.InitialDocsCount
-				ingestedStr = fmt.Sprintf("[green]+%-11s", formatNumber(docChange))
+				ingestedStr = fmt.Sprintf("[green]%-12s", fmt.Sprintf("+%s", formatNumber(docChange)))
 			} else {
-				ingestedStr = fmt.Sprintf("%-12s", "") // Empty space if no changes
+				ingestedStr = fmt.Sprintf("%-12s", "")
 			}
 
 			// Format indexing rate
@@ -879,8 +980,9 @@ func main() {
 			// Convert the size format before display
 			sizeStr := convertSizeFormat(idx.storeSize)
 
-			fmt.Fprintf(indicesPanel, "%s [%s]%-*s  [white][#444444]│[white] %15s [#444444]│[white] %12s [#444444]│[white] %8s [#444444]│[white] %8s [#444444]│[white] %s [#444444]│[white] %-10s\n",
+			fmt.Fprintf(indicesPanel, "%s %s[%s]%-*s[white] [#444444]│[white] %15s [#444444]│[white] %12s [#444444]│[white] %8s [#444444]│[white] %8s [#444444]│[white] %s [#444444]│[white] %-8s\n",
 				writeIcon,
+				streamIndicator,
 				getHealthColor(idx.health),
 				maxIndexNameLen,
 				idx.index,
@@ -1068,6 +1170,9 @@ func main() {
 			case '5':
 				showMetrics = !showMetrics
 				updateGridLayout(grid, showRoles, showIndices, showMetrics)
+			case 'h':
+				showHiddenIndices = !showHiddenIndices
+				// Let the regular update cycle handle it
 			}
 		}
 		return event
@@ -1110,48 +1215,61 @@ func getTotalNetworkRX(stats NodesStats) int64 {
 	return total
 }
 
-func getMaxLengths(nodesInfo NodesInfo, indicesStats IndexStats) (int, int) {
+func getMaxLengths(nodesInfo NodesInfo, indicesStats IndexStats) (int, int, int) {
 	maxNodeNameLen := 0
 	maxIndexNameLen := 0
+	maxTransportLen := 0
 
-	// Get max node name length
+	// Get max node name and transport address length
 	for _, nodeInfo := range nodesInfo.Nodes {
 		if len(nodeInfo.Name) > maxNodeNameLen {
 			maxNodeNameLen = len(nodeInfo.Name)
 		}
-	}
-
-	// Get max index name length
-	for _, index := range indicesStats {
-		if !strings.HasPrefix(index.Index, ".") && // Skip hidden indices
-			len(index.Index) > maxIndexNameLen {
-			maxIndexNameLen = len(index.Index)
+		if len(nodeInfo.TransportAddress) > maxTransportLen {
+			maxTransportLen = len(nodeInfo.TransportAddress)
 		}
 	}
 
-	maxNodeNameLen += 2
-	maxIndexNameLen += 2
+	// Get max index name length only for visible indices
+	for _, index := range indicesStats {
+		// Only consider indices that should be visible based on showHiddenIndices
+		if (showHiddenIndices || !strings.HasPrefix(index.Index, ".")) && index.DocsCount != "0" {
+			if len(index.Index) > maxIndexNameLen {
+				maxIndexNameLen = len(index.Index)
+			}
+		}
+	}
 
-	return maxNodeNameLen, maxIndexNameLen
+	// Add padding
+	maxNodeNameLen += 2
+	maxIndexNameLen += 1 // Single space before separator
+	maxTransportLen += 2 // Add some padding for transport address
+
+	return maxNodeNameLen, maxIndexNameLen, maxTransportLen
 }
 
-func getNodesPanelHeader(maxNodeNameLen int) string {
-	return fmt.Sprintf("[::b]%-*s  [#444444]│[#00ffff] %-13s [#444444]│[#00ffff] %-20s [#444444]│[#00ffff] %-7s [#444444]│[#00ffff] %4s      [#444444]│[#00ffff] %4s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %-25s[white]\n",
+func getNodesPanelHeader(maxNodeNameLen, maxTransportLen int) string {
+	return fmt.Sprintf("[::b]%-*s  [#444444]│[#00ffff] %-13s [#444444]│[#00ffff] %-*s [#444444]│[#00ffff] %-7s [#444444]│[#00ffff] %4s      [#444444]│[#00ffff] %4s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %-16s [#444444]│[#00ffff] %6s [#444444]│[#00ffff] %5s [#444444]│[#00ffff] %8s [#444444]│[#00ffff] %9s [#444444]│[#00ffff] %-25s[white]\n",
 		maxNodeNameLen,
 		"Node Name",
 		"Roles",
+		maxTransportLen,
 		"Transport Address",
 		"Version",
 		"CPU",
 		"Load",
 		"Memory",
 		"Heap",
-		"Disk ",
+		"Disk",
+		"Active",
+		"Queue",
+		"Rejected",
+		"Completed",
 		"OS")
 }
 
 func getIndicesPanelHeader(maxIndexNameLen int) string {
-	return fmt.Sprintf("   [::b]%-*s  [#444444]│[#00ffff] %15s [#444444]│[#00ffff] %12s [#444444]│[#00ffff] %8s [#444444]│[#00ffff] %8s [#444444]│[#00ffff] %-12s [#444444]│[#00ffff] %-10s[white]\n",
+	return fmt.Sprintf("   [::b] %-*s [#444444]│[#00ffff] %15s [#444444]│[#00ffff] %12s [#444444]│[#00ffff] %8s [#444444]│[#00ffff] %8s [#444444]│[#00ffff] %-12s [#444444]│[#00ffff] %-8s[white]\n",
 		maxIndexNameLen,
 		"Index Name",
 		"Documents",
@@ -1161,3 +1279,22 @@ func getIndicesPanelHeader(maxIndexNameLen int) string {
 		"Ingested",
 		"Rate")
 }
+
+func isDataStream(name string, dataStreams DataStreamResponse) bool {
+	for _, ds := range dataStreams.DataStreams {
+		if ds.Name == name {
+			return true
+		}
+	}
+	return false
+}
+
+// Add this with the other type definitions near the top of the file
+type ThreadPoolStats struct {
+	NodeName  string `json:"node_name"`
+	Name      string `json:"name"`
+	Active    string `json:"active"`
+	Queue     string `json:"queue"`
+	Rejected  string `json:"rejected"`
+	Completed string `json:"completed"`
+}