Improve cross-rack replication checks (#138) (#140)

* Use `/bin/bash` for `set_up_net_alias.sh`

`[[ … ]]` is a bashism and doesn’t work with all `sh`.

* Improve `cross-rack` replication checks (#138)

If `--validate-only` is set the number of racks needed is ignored.

If there are broker connections, use the actual number of racks in the
cluster.

Author: joewreschnig

cmd/topicctl/subcmd/check.go

@@ -128,6 +128,8 @@ func checkTopicFile(
 
 	var adminClient admin.Client
 
+	numRacks := -1
+
 	if !checkConfig.validateOnly {
 		var ok bool
 		adminClient, ok = adminClients[clusterConfigPath]
@@ -143,6 +145,10 @@ func checkTopicFile(
 				return false, err
 			}
 			adminClients[clusterConfigPath] = adminClient
+			numRacks, err = countRacks(ctx, adminClient)
+			if err != nil {
+				return false, err
+			}
 		}
 	}
 
@@ -161,10 +167,9 @@ func checkTopicFile(
 			AdminClient:   adminClient,
 			CheckLeaders:  checkConfig.checkLeaders,
 			ClusterConfig: clusterConfig,
-			// TODO: Add support for broker rack verification.
-			NumRacks:     -1,
-			TopicConfig:  topicConfig,
-			ValidateOnly: checkConfig.validateOnly,
+			NumRacks:      numRacks,
+			TopicConfig:   topicConfig,
+			ValidateOnly:  checkConfig.validateOnly,
 		}
 		result, err := cliRunner.CheckTopic(
 			ctx,
@@ -191,3 +196,19 @@ func clusterConfigForTopicCheck(topicConfigPath string) (string, error) {
 		),
 	)
 }
+
+func countRacks(ctx context.Context, c admin.Client) (int, error) {
+	ids, err := c.GetBrokerIDs(ctx)
+	if err != nil {
+		return 0, err
+	}
+	brokers, err := c.GetBrokers(ctx, ids)
+	if err != nil {
+		return 0, err
+	}
+	racks := make(map[string]struct{}, len(brokers))
+	for i := range brokers {
+		racks[brokers[i].Rack] = struct{}{}
+	}
+	return len(racks), nil
+}

pkg/check/check_test.go

@@ -75,6 +75,7 @@ func TestCheck(t *testing.T) {
 		checkTopicConfig config.TopicConfig
 		expectedResults  map[CheckName]bool
 		validateOnly     bool
+		numRacks         int
 	}
 
 	testCases := []testCase{
@@ -194,6 +195,30 @@ func TestCheck(t *testing.T) {
 				CheckNameLeadersCorrect:           true,
 			},
 		},
+		{
+			description: "too few racks",
+			checkTopicConfig: config.TopicConfig{
+				Meta: topicConfig.Meta,
+				// topicConfig.Spec but CrossRack
+				Spec: config.TopicSpec{
+					Partitions:        9,
+					ReplicationFactor: 2,
+					RetentionMinutes:  500,
+					PlacementConfig: config.TopicPlacementConfig{
+						Strategy: config.PlacementStrategyCrossRack,
+						Picker:   config.PickerMethodLowestIndex,
+					},
+					MigrationConfig: &config.TopicMigrationConfig{
+						ThrottleMB:         2,
+						PartitionBatchSize: 3,
+					},
+				},
+			},
+			expectedResults: map[CheckName]bool{
+				CheckNameConfigCorrect: false,
+			},
+			numRacks: 1, // ReplicationFactor - 1
+		},
 	}
 
 	for _, testCase := range testCases {
@@ -203,7 +228,7 @@ func TestCheck(t *testing.T) {
 				AdminClient:   adminClient,
 				ClusterConfig: clusterConfig,
 				CheckLeaders:  true,
-				NumRacks:      -1,
+				NumRacks:      testCase.numRacks,
 				TopicConfig:   testCase.checkTopicConfig,
 				ValidateOnly:  testCase.validateOnly,
 			},

pkg/config/topic.go

@@ -274,7 +274,7 @@ func (t TopicConfig) Validate(numRacks int) error {
 			)
 		}
 	case PlacementStrategyCrossRack:
-		if t.Spec.ReplicationFactor > numRacks {
+		if numRacks > 0 && t.Spec.ReplicationFactor > numRacks {
 			err = multierror.Append(
 				err,
 				fmt.Errorf(

scripts/set_up_net_alias.sh

@@ -1,4 +1,4 @@
-#!/bin/sh
+#!/bin/bash
 
 ADDR=169.254.123.123
 echo "Aliasing $ADDR to localhost..."