Merge pull request #222 from DataDog/jamie/rebalance-metrics

rebalance checks broker metrics

cmd/topicmappr/commands/config.go

@@ -98,6 +98,21 @@ func initZooKeeper(cmd *cobra.Command) (kafkazk.Handler, error) {
 	return zk, nil
 }
 
+// ensureBrokerMetrics takes a map of reference brokers and
+// a map of discovered broker metadata. Any non-missing brokers
+// in the broker map must be present in the broker metadata map
+// and have a non-true MetricsIncomplete value.
+func ensureBrokerMetrics(cmd *cobra.Command, bm kafkazk.BrokerMap, bmm kafkazk.BrokerMetaMap) {
+	for id, b := range bm {
+		// Missing brokers won't even
+		// be found in the brokerMeta.
+		if !b.Missing && id != 0 && bmm[id].MetricsIncomplete {
+			fmt.Printf("Metrics not found for broker %d\n", id)
+			os.Exit(1)
+		}
+	}
+}
+
 // containsRegex takes a topic name
 // reference and returns whether or not
 // it should be interpreted as regex.

cmd/topicmappr/commands/rebalance_steps.go

@@ -102,6 +102,10 @@ func validateBrokersForRebalance(cmd *cobra.Command, brokers kafkazk.BrokerMap,
 		fmt.Printf("%s-\n", indent)
 	}
 
+	// Check if any referenced brokers are marked as having
+	// missing/partial metrics data.
+	ensureBrokerMetrics(cmd, brokers, bm)
+
 	switch {
 	case c.Missing > 0, c.OldMissing > 0, c.Replace > 0:
 		fmt.Printf("%s[ERROR] rebalance only allows broker additions\n", indent)

cmd/topicmappr/commands/rebuild.go

@@ -139,7 +139,9 @@ func rebuild(cmd *cobra.Command, _ []string) {
 
 	// Check if any referenced brokers are marked as having
 	// missing/partial metrics data.
-	ensureBrokerMetrics(cmd, brokers, brokerMeta)
+	if m, _ := cmd.Flags().GetBool("use-meta"); m {
+		ensureBrokerMetrics(cmd, brokers, brokerMeta)
+	}
 
 	// Create substitution affinities.
 	affinities := getSubAffinities(cmd, brokers, brokersOrig, partitionMapIn)

cmd/topicmappr/commands/rebuild_steps.go

@@ -45,23 +45,6 @@ func getPartitionMap(cmd *cobra.Command, zk kafkazk.Handler) *kafkazk.PartitionM
 	return nil
 }
 
-// ensureBrokerMetrics takes a map of reference brokers and
-// a map of discovered broker metadata. Any non-missing brokers
-// in the broker map must be present in the broker metadata map
-// and have a non-true MetricsIncomplete value.
-func ensureBrokerMetrics(cmd *cobra.Command, bm kafkazk.BrokerMap, bmm kafkazk.BrokerMetaMap) {
-	if m, _ := cmd.Flags().GetBool("use-meta"); m {
-		for id, b := range bm {
-			// Missing brokers won't even
-			// be found in the brokerMeta.
-			if !b.Missing && id != 0 && bmm[id].MetricsIncomplete {
-				fmt.Printf("Metrics not found for broker %d\n", id)
-				os.Exit(1)
-			}
-		}
-	}
-}
-
 // getSubAffinities, if enabled via --sub-affinity, takes reference broker maps
 // and a partition map and attempts to return a complete SubstitutionAffinities.
 func getSubAffinities(cmd *cobra.Command, bm kafkazk.BrokerMap, bmo kafkazk.BrokerMap, pm *kafkazk.PartitionMap) kafkazk.SubstitutionAffinities {