refactor: remove dead REST-oriented service methods after RPC migration (UX-1199)

Remove CreateACL, DeleteACLs, DescribeQuotas methods and their
associated types (DeleteACLsResponse, QuotaResponse*) from the console
Servicer interface and Service implementation. These only served the
deleted REST handlers and are fully replaced by the Kafka proxy methods
(CreateACLs, DeleteACLsKafka, DescribeClientQuotas) used by Connect RPC.

Author: c-julin

backend/pkg/console/create_acl.go

@@ -11,61 +11,10 @@ package console
 
 import (
 	"context"
-	"errors"
-	"fmt"
-	"log/slog"
-	"net/http"
 
-	"github.com/cloudhut/common/rest"
-	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kmsg"
 )
 
-// CreateACL creates an ACL resource in your target Kafka cluster.
-func (s *Service) CreateACL(ctx context.Context, createReq kmsg.CreateACLsRequestCreation) *rest.Error {
-	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)
-	if err != nil {
-		return errorToRestError(err)
-	}
-
-	req := kmsg.NewCreateACLsRequest()
-	req.Creations = []kmsg.CreateACLsRequestCreation{createReq}
-	res, err := req.RequestWith(ctx, cl)
-	if err != nil {
-		return &rest.Error{
-			Err:          err,
-			Status:       http.StatusServiceUnavailable,
-			Message:      fmt.Sprintf("Failed to execute create ACL command: %v", err.Error()),
-			InternalLogs: []slog.Attr{slog.Any("create_acl_req", createReq)},
-			IsSilent:     false,
-		}
-	}
-
-	if len(res.Results) != 1 {
-		return &rest.Error{
-			Err:          errors.New("unexpected number of results in create ACL response"),
-			Status:       http.StatusInternalServerError,
-			Message:      fmt.Sprintf("Failed to execute delete topic command: %v", err.Error()),
-			InternalLogs: []slog.Attr{slog.Int("results_length", len(res.Results))},
-			IsSilent:     false,
-		}
-	}
-
-	aclRes := res.Results[0]
-	err = kerr.ErrorForCode(aclRes.ErrorCode)
-	if err != nil {
-		return &rest.Error{
-			Err:          err,
-			Status:       http.StatusServiceUnavailable,
-			Message:      fmt.Sprintf("Failed to execute create ACL command: %v", err.Error()),
-			InternalLogs: []slog.Attr{slog.Any("create_acl_req", createReq)},
-			IsSilent:     false,
-		}
-	}
-
-	return nil
-}
-
 // CreateACLs proxies the request/response to CreateACLs via the Kafka API.
 func (s *Service) CreateACLs(ctx context.Context, req *kmsg.CreateACLsRequest) (*kmsg.CreateACLsResponse, error) {
 	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)

backend/pkg/console/delete_acls.go

@@ -11,73 +11,10 @@ package console
 
 import (
 	"context"
-	"fmt"
-	"log/slog"
-	"net/http"
 
-	"github.com/cloudhut/common/rest"
 	"github.com/twmb/franz-go/pkg/kmsg"
 )
 
-// DeleteACLsResponse is the response to deleting ACL resources.
-type DeleteACLsResponse struct {
-	ErrorMessages []string `json:"errorMessage"`
-	MatchedACLs   int      `json:"matchedACLs"`
-	DeletedACLs   int      `json:"deletedACLs"`
-}
-
-// DeleteACLs deletes Kafka ACLs based on a given filter.
-func (s *Service) DeleteACLs(ctx context.Context, filter kmsg.DeleteACLsRequestFilter) (DeleteACLsResponse, *rest.Error) {
-	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)
-	if err != nil {
-		return DeleteACLsResponse{}, errorToRestError(err)
-	}
-
-	req := kmsg.NewDeleteACLsRequest()
-	req.Filters = []kmsg.DeleteACLsRequestFilter{filter}
-
-	res, err := req.RequestWith(ctx, cl)
-	if err != nil {
-		return DeleteACLsResponse{}, &rest.Error{
-			Err:          err,
-			Status:       http.StatusServiceUnavailable,
-			Message:      fmt.Sprintf("Failed to execute delete topic command: %v", err.Error()),
-			InternalLogs: []slog.Attr{slog.Any("delete_acl_req", filter)},
-			IsSilent:     false,
-		}
-	}
-
-	deleteAclsRes := DeleteACLsResponse{
-		ErrorMessages: make([]string, 0),
-		MatchedACLs:   0,
-		DeletedACLs:   0,
-	}
-	for _, aclRes := range res.Results {
-		err := newKafkaErrorWithDynamicMessage(aclRes.ErrorCode, aclRes.ErrorMessage)
-		if err != nil {
-			return DeleteACLsResponse{}, &rest.Error{
-				Err:          err,
-				Status:       http.StatusServiceUnavailable,
-				Message:      fmt.Sprintf("Failed to delete Kafka ACL: %v", err.Error()),
-				InternalLogs: []slog.Attr{slog.Any("delete_acl_req", filter)},
-				IsSilent:     false,
-			}
-		}
-
-		for _, item := range aclRes.MatchingACLs {
-			deleteAclsRes.MatchedACLs++
-			err := newKafkaErrorWithDynamicMessage(item.ErrorCode, item.ErrorMessage)
-			if err != nil {
-				deleteAclsRes.ErrorMessages = append(deleteAclsRes.ErrorMessages, err.Error())
-				continue
-			}
-			deleteAclsRes.DeletedACLs++
-		}
-	}
-
-	return deleteAclsRes, nil
-}
-
 // DeleteACLsKafka proxies the request/response via the Kafka API.
 func (s *Service) DeleteACLsKafka(ctx context.Context, req *kmsg.DeleteACLsRequest) (*kmsg.DeleteACLsResponse, error) {
 	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)

backend/pkg/console/describe_quotas.go

@@ -11,89 +11,10 @@ package console
 
 import (
 	"context"
-	"fmt"
 
-	"github.com/twmb/franz-go/pkg/kerr"
 	"github.com/twmb/franz-go/pkg/kmsg"
 )
 
-// QuotaResponse is a helper type that carries the sum of all quota responses
-// sent by each broker in the cluster.
-type QuotaResponse struct {
-	Error string              `json:"error,omitempty"`
-	Items []QuotaResponseItem `json:"items"`
-}
-
-// QuotaResponseItem is a broker's response to a quota entity in Kafka.
-type QuotaResponseItem struct {
-	EntityType string                 `json:"entityType"`
-	EntityName string                 `json:"entityName"`
-	Settings   []QuotaResponseSetting `json:"settings"`
-}
-
-// QuotaResponseSetting is a quota configuration.
-type QuotaResponseSetting struct {
-	Key   string  `json:"key"`
-	Value float64 `json:"value"`
-}
-
-// DescribeQuotas fetches the configured quota settings in the target cluster.
-func (s *Service) DescribeQuotas(ctx context.Context) QuotaResponse {
-	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)
-	if err != nil {
-		return QuotaResponse{
-			Error: err.Error(),
-		}
-	}
-
-	items := make([]QuotaResponseItem, 0)
-	req := kmsg.NewDescribeClientQuotasRequest()
-	quotas, err := req.RequestWith(ctx, cl)
-	if err != nil {
-		return QuotaResponse{
-			Error: fmt.Errorf("kafka request has failed: %w", err).Error(),
-			Items: nil,
-		}
-	}
-
-	err = kerr.ErrorForCode(quotas.ErrorCode)
-	if err != nil {
-		return QuotaResponse{
-			Error: fmt.Errorf("inner kafka error: %w", err).Error(),
-			Items: nil,
-		}
-	}
-
-	// Flat map all quota settings from response into our items array
-	for _, entry := range quotas.Entries {
-		settings := make([]QuotaResponseSetting, len(entry.Values))
-		for i, setting := range entry.Values {
-			settings[i] = QuotaResponseSetting{
-				Key:   setting.Key,
-				Value: setting.Value,
-			}
-		}
-
-		for _, entity := range entry.Entity {
-			// A nil value for entity.Name means that this quota is the default for the respective entity.Type
-			entityName := "<default>"
-			if entity.Name != nil {
-				entityName = *entity.Name
-			}
-			items = append(items, QuotaResponseItem{
-				EntityType: entity.Type,
-				EntityName: entityName,
-				Settings:   settings,
-			})
-		}
-	}
-
-	return QuotaResponse{
-		Error: "",
-		Items: items,
-	}
-}
-
 // DescribeClientQuotas proxies the request/response for describing client quotas via the Kafka API.
 func (s *Service) DescribeClientQuotas(ctx context.Context, req *kmsg.DescribeClientQuotasRequest) (*kmsg.DescribeClientQuotasResponse, error) {
 	cl, _, err := s.kafkaClientFactory.GetKafkaClient(ctx)

backend/pkg/console/servicer.go

@@ -25,13 +25,10 @@ type Servicer interface {
 	GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
 	DeleteConsumerGroup(ctx context.Context, groupID string) error
 	GetConsumerGroupsOverview(ctx context.Context, groupIDs []string) ([]ConsumerGroupOverview, *rest.Error)
-	CreateACL(ctx context.Context, createReq kmsg.CreateACLsRequestCreation) *rest.Error
 	CreateTopic(ctx context.Context, createTopicReq kmsg.CreateTopicsRequestTopic) (CreateTopicResponse, *rest.Error)
-	DeleteACLs(ctx context.Context, filter kmsg.DeleteACLsRequestFilter) (DeleteACLsResponse, *rest.Error)
 	DeleteConsumerGroupOffsets(ctx context.Context, groupID string, topics []kmsg.OffsetDeleteRequestTopic) ([]DeleteConsumerGroupOffsetsResponseTopic, error)
 	DeleteTopic(ctx context.Context, topicName string) *rest.Error
 	DeleteTopicRecords(ctx context.Context, deleteReq kmsg.DeleteRecordsRequestTopic) (DeleteTopicRecordsResponse, *rest.Error)
-	DescribeQuotas(ctx context.Context) QuotaResponse
 	EditConsumerGroupOffsets(ctx context.Context, groupID string, topics []kmsg.OffsetCommitRequestTopic) (*EditConsumerGroupOffsetsResponse, *rest.Error)
 	EditTopicConfig(ctx context.Context, topicName string, configs []kmsg.IncrementalAlterConfigsRequestResourceConfig) error
 	GetEndpointCompatibility(ctx context.Context) (EndpointCompatibility, error)