cmq -- Cherami Metadata Query-er

Makefile

@@ -96,7 +96,10 @@ cherami-store-tool: $(DEPS)
 cdb: $(DEPS)
 	go build -i $(EMBED) -o cdb cmd/tools/cdb/*.go
 
-bins: cherami-server cherami-replicator-server cherami-cli cherami-admin cherami-replicator-tool cherami-cassandra-tool cherami-store-tool cdb
+cmq: $(DEPS)
+	go build -i $(EMBED) -o cmq cmd/tools/cmq/*.go
+
+bins: cherami-server cherami-replicator-server cherami-cli cherami-admin cherami-replicator-tool cherami-cassandra-tool cherami-store-tool cdb cmq
 
 cover_profile: lint bins
 	@mkdir -p $(BUILD)
@@ -123,7 +126,7 @@ cover_ci: cover_profile
 	goveralls -coverprofile=$(BUILD)/cover.out -service=travis-ci || echo -e "\x1b[31mCoveralls failed\x1b[m"
 
 clean:
-	rm -f cherami-server cherami-replicator-server cherami-cli cherami-admin cherami-replicator-tool cherami-cassandra-tool cherami-store-tool cdb
+	rm -f cherami-server cherami-replicator-server cherami-cli cherami-admin cherami-replicator-tool cherami-cassandra-tool cherami-store-tool cdb cmq
 	rm -Rf vendor/*
 	rm -Rf $(BUILD)
 

clients/metadata/metadata_cassandra.go

@@ -198,18 +198,18 @@ func parseConsistency(cfgCons string) (lowCons gocql.Consistency, midCons gocql.
 
 	switch cons := strings.Split(cfgCons, ","); len(cons) {
 	case 3:
-		lowCons = gocql.ParseConsistency(strings.TrimSpace(cons[2]))
+		lowCons, _ = gocql.ParseConsistency(strings.TrimSpace(cons[2]))
 		fallthrough
 
 	case 2:
-		midCons = gocql.ParseConsistency(strings.TrimSpace(cons[1]))
+		midCons, _ = gocql.ParseConsistency(strings.TrimSpace(cons[1]))
 		if len(cons) == 2 {
 			lowCons = midCons
 		}
 		fallthrough
 
 	case 1:
-		highCons = gocql.ParseConsistency(strings.TrimSpace(cons[0]))
+		highCons, _ = gocql.ParseConsistency(strings.TrimSpace(cons[0]))
 	}
 
 	return
@@ -1189,12 +1189,29 @@ const (
 		columnZoneConfigs + `: ?, ` +
 		columnOptions + `: ? }`
 
+	sqlConsumerGroupType = columnConsumerGroup + `.` + columnUUID + "," +
+		columnConsumerGroup + `.` + columnDestinationUUID + "," +
+		columnConsumerGroup + `.` + columnName + "," +
+		columnConsumerGroup + `.` + columnStartFrom + "," +
+		columnConsumerGroup + `.` + columnStatus + "," +
+		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
+		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
+		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
+		columnConsumerGroup + `.` + columnDelaySeconds + "," +
+		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
+		columnConsumerGroup + `.` + columnOwnerEmail + "," +
+		columnConsumerGroup + `.` + columnIsMultiZone + "," +
+		columnConsumerGroup + `.` + columnActiveZone + "," +
+		columnConsumerGroup + `.` + columnZoneConfigs + "," +
+		columnConsumerGroup + `.` + columnOptions
+
 	sqlInsertCGByUUID = `INSERT INTO ` + tableConsumerGroups +
 		`(` +
 		columnUUID + `, ` +
+		columnDestinationUUID + `, ` +
 		columnIsMultiZone + `, ` +
 		columnConsumerGroup +
-		`) VALUES (?, ?,` + sqlCGValue + `)`
+		`) VALUES (?, ?, ?, ` + sqlCGValue + `)`
 
 	sqlInsertCGByName = `INSERT INTO ` + tableConsumerGroupsByName +
 		`(` +
@@ -1205,65 +1222,28 @@ const (
 		`) VALUES (?, ?, ?, ` + sqlCGValue + `) IF NOT EXISTS`
 
 	sqlGetCGByName = `SELECT  ` +
-		columnConsumerGroup + `.` + columnUUID + "," +
-		columnDestinationUUID + "," +
-		columnName + "," +
-		columnConsumerGroup + `.` + columnStartFrom + "," +
-		columnConsumerGroup + `.` + columnStatus + "," +
-		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
-		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
-		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
-		columnConsumerGroup + `.` + columnDelaySeconds + "," +
-		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
-		columnConsumerGroup + `.` + columnOwnerEmail + "," +
-		columnConsumerGroup + `.` + columnIsMultiZone + "," +
-		columnConsumerGroup + `.` + columnActiveZone + "," +
-		columnConsumerGroup + `.` + columnZoneConfigs + "," +
-		columnConsumerGroup + `.` + columnOptions +
+		sqlConsumerGroupType +
 		` FROM ` + tableConsumerGroupsByName +
 		` WHERE ` + columnDestinationUUID + `=? and ` + columnName + `=?`
 
 	sqlGetCG = `SELECT  ` +
-		columnConsumerGroup + `.` + columnUUID + "," +
-		columnConsumerGroup + `.` + columnDestinationUUID + "," +
-		columnConsumerGroup + `.` + columnName + "," +
-		columnConsumerGroup + `.` + columnStartFrom + "," +
-		columnConsumerGroup + `.` + columnStatus + "," +
-		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
-		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
-		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
-		columnConsumerGroup + `.` + columnDelaySeconds + "," +
-		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
-		columnConsumerGroup + `.` + columnOwnerEmail + "," +
-		columnConsumerGroup + `.` + columnIsMultiZone + "," +
-		columnConsumerGroup + `.` + columnActiveZone + "," +
-		columnConsumerGroup + `.` + columnZoneConfigs + "," +
-		columnConsumerGroup + `.` + columnOptions +
+		sqlConsumerGroupType +
 		` FROM ` + tableConsumerGroups
 
 	sqlGetCGByUUID = sqlGetCG + ` WHERE ` + columnUUID + `=?`
 
 	sqlListCGsByDestUUID = `SELECT  ` +
-		columnConsumerGroup + `.` + columnUUID + "," +
-		columnDestinationUUID + "," +
-		columnName + "," +
-		columnConsumerGroup + `.` + columnStartFrom + "," +
-		columnConsumerGroup + `.` + columnStatus + "," +
-		columnConsumerGroup + `.` + columnLockTimeoutSeconds + "," +
-		columnConsumerGroup + `.` + columnMaxDeliveryCount + "," +
-		columnConsumerGroup + `.` + columnSkipOlderMessagesSeconds + "," +
-		columnConsumerGroup + `.` + columnDelaySeconds + "," +
-		columnConsumerGroup + `.` + columnDeadLetterQueueDestinationUUID + "," +
-		columnConsumerGroup + `.` + columnOwnerEmail + "," +
-		columnConsumerGroup + `.` + columnIsMultiZone + "," +
-		columnConsumerGroup + `.` + columnActiveZone + "," +
-		columnConsumerGroup + `.` + columnZoneConfigs + "," +
-		columnConsumerGroup + `.` + columnOptions +
+		sqlConsumerGroupType +
 		` FROM ` + tableConsumerGroupsByName +
 		` WHERE ` + columnDestinationUUID + `=?`
 
+	sqlListCGsUUID = `SELECT  ` +
+		sqlConsumerGroupType +
+		` FROM ` + tableConsumerGroups +
+		` WHERE ` + columnDestinationUUID + `=?`
+
 	sqlUpdateCGByUUID = `UPDATE ` + tableConsumerGroups +
-		` SET ` + columnIsMultiZone + ` = ?, ` + columnConsumerGroup + `= ` + sqlCGValue +
+		` SET ` + columnDestinationUUID + ` = ?, ` + columnIsMultiZone + ` = ?, ` + columnConsumerGroup + `= ` + sqlCGValue +
 		` WHERE ` + columnUUID + `=?`
 
 	sqlUpdateCGByName = `UPDATE ` + tableConsumerGroupsByName +
@@ -1334,6 +1314,7 @@ func (s *CassandraMetadataService) CreateConsumerGroupUUID(ctx thrift.Context, r
 
 	err = s.session.Query(sqlInsertCGByUUID,
 		cgUUID,
+		dstUUID,
 		createRequest.GetIsMultiZone(),
 		cgUUID,
 		dstUUID,
@@ -1651,6 +1632,7 @@ func (s *CassandraMetadataService) UpdateConsumerGroup(ctx thrift.Context, reque
 
 	batch.Query(sqlUpdateCGByUUID,
 		// Value columns
+		newCG.GetDestinationUUID(),
 		newCG.GetIsMultiZone(),
 		newCG.GetConsumerGroupUUID(),
 		newCG.GetDestinationUUID(),
@@ -1794,6 +1776,7 @@ func (s *CassandraMetadataService) DeleteConsumerGroup(ctx thrift.Context, reque
 
 	batch.Query(sqlUpdateCGByUUID,
 		// Value columns
+		existingCG.GetDestinationUUID(),
 		existingCG.GetIsMultiZone(),
 		existingCG.GetConsumerGroupUUID(),
 		existingCG.GetDestinationUUID(),
@@ -1862,6 +1845,7 @@ func (s *CassandraMetadataService) DeleteConsumerGroupUUID(ctx thrift.Context, r
 
 	query := s.session.Query(sqlDeleteCGByUUIDWithTTL,
 		existing.GetConsumerGroupUUID(),
+		existing.GetDestinationUUID(),
 		existing.GetIsMultiZone(),
 		existing.GetConsumerGroupUUID(),
 		existing.GetDestinationUUID(),
@@ -2003,6 +1987,73 @@ func (s *CassandraMetadataService) ListConsumerGroups(ctx thrift.Context, reques
 	return result, nil
 }
 
+// ListConsumerGroupsUUID returns all ConsumerGroups matching the given destination-uuid.
+func (s *CassandraMetadataService) ListConsumerGroupsUUID(ctx thrift.Context, request *shared.ListConsumerGroupsUUIDRequest) (*shared.ListConsumerGroupsUUIDResult_, error) {
+
+	if !request.IsSetDestinationUUID() {
+		return nil, &shared.BadRequestError{
+			Message: fmt.Sprintf("DestinationUUID not specified"),
+		}
+	}
+
+	dstUUID := request.GetDestinationUUID()
+
+	var iter *gocql.Iter
+	iter = s.session.Query(sqlListCGsUUID, dstUUID).Consistency(s.lowConsLevel).PageSize(int(request.GetLimit())).PageState(request.PageToken).Iter()
+
+	if iter == nil {
+		return nil, &shared.InternalServiceError{
+			Message: "Query returned nil iterator",
+		}
+	}
+
+	result := &shared.ListConsumerGroupsUUIDResult_{
+		ConsumerGroups: []*shared.ConsumerGroupDescription{},
+		NextPageToken:  request.PageToken,
+	}
+	cg := getUtilConsumerGroupDescription()
+	var zoneConfigsData []map[string]interface{}
+	for iter.Scan(
+		cg.ConsumerGroupUUID,
+		cg.DestinationUUID,
+		cg.ConsumerGroupName,
+		cg.StartFrom,
+		cg.Status,
+		cg.LockTimeoutSeconds,
+		cg.MaxDeliveryCount,
+		cg.SkipOlderMessagesSeconds,
+		cg.DelaySeconds,
+		&cg.DeadLetterQueueDestinationUUID,
+		cg.OwnerEmail,
+		cg.IsMultiZone,
+		cg.ActiveZone,
+		&zoneConfigsData,
+		&cg.Options) {
+
+		// skip over deleted rows
+		if cg.GetStatus() == shared.ConsumerGroupStatus_DELETED {
+			zoneConfigsData = nil
+			continue
+		}
+
+		cg.ZoneConfigs = unmarshalCgZoneConfigs(zoneConfigsData)
+		result.ConsumerGroups = append(result.ConsumerGroups, cg)
+		cg = getUtilConsumerGroupDescription()
+		zoneConfigsData = nil
+	}
+
+	nextPageToken := iter.PageState()
+	result.NextPageToken = make([]byte, len(nextPageToken))
+	copy(result.NextPageToken, nextPageToken)
+	if err := iter.Close(); err != nil {
+		return nil, &shared.InternalServiceError{
+			Message: err.Error(),
+		}
+	}
+
+	return result, nil
+}
+
 // ListAllConsumerGroups returns all ConsumerGroups in ConsumerGroups Table. This API is only used for debuging tool
 func (s *CassandraMetadataService) ListAllConsumerGroups(ctx thrift.Context, request *shared.ListConsumerGroupRequest) (*shared.ListConsumerGroupResult_, error) {
 
@@ -3394,7 +3445,10 @@ func (s *CassandraMetadataService) UpdateStoreExtentReplicaStats(ctx thrift.Cont
 		)
 	}
 	if err := s.session.ExecuteBatch(batch); err != nil {
-		s.log.WithField(common.TagExt, request.GetExtentUUID()).Error("UpdateExtentReplicaStats failed")
+		s.log.WithFields(bark.Fields{
+			common.TagExt: request.GetExtentUUID(),
+			common.TagErr: err,
+		}).Error("UpdateExtentReplicaStats failed")
 		return &shared.InternalServiceError{
 			Message: "UpdateStoreExtentReplicaStats: %v" + err.Error(),
 		}