Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 103 additions & 0 deletions backend/pkg/api/connect/service/topic/v1/console.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
// Copyright 2023 Redpanda Data, Inc.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.md
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0

package topic

import (
"context"
"fmt"
"log/slog"
"sort"
"strings"

"connectrpc.com/connect"
"github.com/redpanda-data/common-go/api/pagination"

apierrors "github.com/redpanda-data/console/backend/pkg/api/connect/errors"
"github.com/redpanda-data/console/backend/pkg/console"
v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
"github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1/consolev1alpha1connect"
v1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/dataplane/v1"
)

var _ consolev1alpha1connect.TopicServiceHandler = (*ConsoleService)(nil)

// ConsoleService is the Console-facing implementation of the topic service.
//
// Unlike the other Console services it does not wrap the dataplane handler: the dataplane
// ListTopics deliberately returns only cheap metadata, whereas the Console topic list also renders
// cleanup policy and on-disk size. Those come from console.GetTopicsOverview, which fans out a
// DescribeLogDirs request to every broker — a cost the public dataplane API should not pay.
type ConsoleService struct {
logger *slog.Logger
consoleSvc console.Servicer
mapper consoleMapper
defaulter defaulter
}

// NewConsoleService creates a new ConsoleService.
func NewConsoleService(logger *slog.Logger, consoleSvc console.Servicer) *ConsoleService {
return &ConsoleService{
logger: logger,
consoleSvc: consoleSvc,
mapper: consoleMapper{},
defaulter: defaulter{},
}
}

// ListTopics lists Kafka topics along with the metadata the Console topic list renders.
func (s *ConsoleService) ListTopics(ctx context.Context, req *connect.Request[v1alpha1.ListTopicsRequest]) (*connect.Response[v1alpha1.ListTopicsResponse], error) {
dataplaneReq := req.Msg.GetRequest()
if dataplaneReq == nil {
dataplaneReq = &v1.ListTopicsRequest{}
}
s.defaulter.applyListTopicsRequest(dataplaneReq)

topicSummaries, err := s.consoleSvc.GetTopicsOverview(ctx)
if err != nil {
return nil, apierrors.NewConnectError(
connect.CodeInternal,
err,
apierrors.NewErrorInfo(v1.Reason_REASON_KAFKA_API_ERROR.String(), apierrors.KeyValsFromKafkaError(err)...),
)
}

if nameContains := dataplaneReq.GetFilter().GetNameContains(); nameContains != "" {
filteredTopics := make([]*console.TopicSummary, 0, len(topicSummaries))
for _, topic := range topicSummaries {
if strings.Contains(topic.TopicName, nameContains) {
filteredTopics = append(filteredTopics, topic)
}
}
topicSummaries = filteredTopics
}

topics := s.mapper.topicSummariesToProto(topicSummaries)

var nextPageToken string
if dataplaneReq.GetPageSize() > 0 {
sort.SliceStable(topics, func(i, j int) bool {
return topics[i].GetName() < topics[j].GetName()
})
page, token, err := pagination.SliceToPaginatedWithToken(topics, int(dataplaneReq.GetPageSize()), dataplaneReq.GetPageToken(), "name", func(x *v1alpha1.ListTopicsResponse_Topic) string {
return x.GetName()
})
if err != nil {
return nil, apierrors.NewConnectError(
connect.CodeInternal,
fmt.Errorf("failed to apply pagination: %w", err),
apierrors.NewErrorInfo(v1.Reason_REASON_CONSOLE_ERROR.String()),
)
}
topics = page
nextPageToken = token
}

return connect.NewResponse(&v1alpha1.ListTopicsResponse{Topics: topics, NextPageToken: nextPageToken}), nil
}
54 changes: 54 additions & 0 deletions backend/pkg/api/connect/service/topic/v1/console_mapper.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
// Copyright 2023 Redpanda Data, Inc.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.md
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0

package topic

import (
"github.com/redpanda-data/console/backend/pkg/console"
v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
)

// consoleMapper maps console topic summaries to the Console v1alpha1 proto types.
type consoleMapper struct{}

func (k *consoleMapper) topicSummariesToProto(summaries []*console.TopicSummary) []*v1alpha1.ListTopicsResponse_Topic {
topics := make([]*v1alpha1.ListTopicsResponse_Topic, len(summaries))
for i, summary := range summaries {
topics[i] = k.topicSummaryToProto(summary)
}

return topics
}

func (k *consoleMapper) topicSummaryToProto(summary *console.TopicSummary) *v1alpha1.ListTopicsResponse_Topic {
return &v1alpha1.ListTopicsResponse_Topic{
Name: summary.TopicName,
Internal: summary.IsInternal,
PartitionCount: int32(summary.PartitionCount),
ReplicationFactor: int32(summary.ReplicationFactor),
CleanupPolicy: summary.CleanupPolicy,
LogDirSummary: k.topicLogDirSummaryToProto(summary.LogDirSummary),
}
}

func (*consoleMapper) topicLogDirSummaryToProto(summary console.TopicLogDirSummary) *v1alpha1.ListTopicsResponse_LogDirSummary {
replicaErrors := make([]*v1alpha1.ListTopicsResponse_ReplicaError, len(summary.ReplicaErrors))
for i, replicaError := range summary.ReplicaErrors {
replicaErrors[i] = &v1alpha1.ListTopicsResponse_ReplicaError{
BrokerId: replicaError.BrokerID,
Error: replicaError.Error,
}
}

return &v1alpha1.ListTopicsResponse_LogDirSummary{
TotalSizeBytes: summary.TotalSizeBytes,
ReplicaErrors: replicaErrors,
Hint: summary.Hint,
}
}
101 changes: 101 additions & 0 deletions backend/pkg/api/connect/service/topic/v1/console_mapper_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
// Copyright 2024 Redpanda Data, Inc.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.md
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0

package topic

import (
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

"github.com/redpanda-data/console/backend/pkg/console"
v1alpha1 "github.com/redpanda-data/console/backend/pkg/protogen/redpanda/api/console/v1alpha1"
)

// TestConsoleTopicSummariesToProto guards the Console ListTopics response shape: the cleanup policy
// and the aggregated log dir summary (size, hint, per-broker replica errors) must survive the
// mapping from console.TopicSummary. A regression here surfaces as empty cleanup-policy / size
// columns in the UI. These fields live on the Console API rather than the dataplane one, which
// deliberately exposes only cheap metadata.
func TestConsoleTopicSummariesToProto(t *testing.T) {
testCases := []struct {
name string
input []*console.TopicSummary
validateFn func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic)
}{
{
name: "maps cleanup policy and log dir summary",
input: []*console.TopicSummary{
{
TopicName: "orders",
IsInternal: false,
PartitionCount: 3,
ReplicationFactor: 2,
CleanupPolicy: "compact",
LogDirSummary: console.TopicLogDirSummary{
TotalSizeBytes: 2048,
Hint: "partial result",
ReplicaErrors: []console.TopicLogDirSummaryReplicaError{
{BrokerID: 1, Error: "broker down"},
},
},
},
},
validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
require.Len(t, result, 1)
topic := result[0]
assert.Equal(t, "orders", topic.GetName())
assert.False(t, topic.GetInternal())
assert.Equal(t, int32(3), topic.GetPartitionCount())
assert.Equal(t, int32(2), topic.GetReplicationFactor())
assert.Equal(t, "compact", topic.GetCleanupPolicy())

summary := topic.GetLogDirSummary()
require.NotNil(t, summary)
assert.Equal(t, int64(2048), summary.GetTotalSizeBytes())
assert.Equal(t, "partial result", summary.GetHint())
require.Len(t, summary.GetReplicaErrors(), 1)
assert.Equal(t, int32(1), summary.GetReplicaErrors()[0].GetBrokerId())
assert.Equal(t, "broker down", summary.GetReplicaErrors()[0].GetError())
},
},
{
name: "preserves order and the N/A cleanup policy fallback",
input: []*console.TopicSummary{
{TopicName: "a", CleanupPolicy: "delete", LogDirSummary: console.TopicLogDirSummary{TotalSizeBytes: 1}},
{TopicName: "b", CleanupPolicy: "N/A", LogDirSummary: console.TopicLogDirSummary{TotalSizeBytes: 2}},
},
validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
require.Len(t, result, 2)
assert.Equal(t, "a", result[0].GetName())
assert.Equal(t, "delete", result[0].GetCleanupPolicy())
assert.Equal(t, "b", result[1].GetName())
assert.Equal(t, "N/A", result[1].GetCleanupPolicy())
assert.Empty(t, result[1].GetLogDirSummary().GetReplicaErrors())
},
},
{
name: "empty input yields empty slice",
input: []*console.TopicSummary{},
validateFn: func(t *testing.T, result []*v1alpha1.ListTopicsResponse_Topic) {
assert.Empty(t, result)
},
},
}

consoleMapper := consoleMapper{}

for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
out := consoleMapper.topicSummariesToProto(tc.input)
tc.validateFn(t, out)
})
}
}
10 changes: 10 additions & 0 deletions backend/pkg/api/routes.go
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,7 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
transformSvcV1 := transformsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "transform_service"), v, api.RedpandaClientProvider)
kafkaConnectSvcV1 := apikafkaconnectsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "kafka_connect_service"), api.ConnectSvc)
consoleTransformSvcV1 := &transformsvcv1.ConsoleService{Impl: transformSvcV1}
consoleTopicSvcV1 := topicsvcv1.NewConsoleService(loggerpkg.Named(api.Logger, "topic_service"), api.ConsoleSvc)
monitoringSvcV1 := monitoringsvcv1.NewService(api.Cfg, loggerpkg.Named(api.Logger, "monitoring_service"), api.RedpandaClientProvider)

// v1alpha2
Expand Down Expand Up @@ -170,6 +171,7 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
consolev1alpha1connect.SecurityServiceName: consolev1alpha1connect.UnimplementedSecurityServiceHandler{},
consolev1alpha1connect.LicenseServiceName: licenseSvc,
consolev1alpha1connect.TransformServiceName: consoleTransformSvcV1,
consolev1alpha1connect.TopicServiceName: consoleTopicSvcV1,
consolev1alpha1connect.AuthenticationServiceName: &AuthenticationDefaultHandler{},
consolev1alpha1connect.ClusterStatusServiceName: clusterStatusSvc,
consolev1alpha1connect.SecretServiceName: consolev1alpha1connect.UnimplementedSecretServiceHandler{},
Expand Down Expand Up @@ -247,6 +249,9 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
consoleTransformSvcPath, consoleTransformSvcHandler := consolev1alpha1connect.NewTransformServiceHandler(
hookOutput.Services[consolev1alpha1connect.TransformServiceName].(consolev1alpha1connect.TransformServiceHandler),
connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
consoleTopicSvcPath, consoleTopicSvcHandler := consolev1alpha1connect.NewTopicServiceHandler(
hookOutput.Services[consolev1alpha1connect.TopicServiceName].(consolev1alpha1connect.TopicServiceHandler),
connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
licenseSvcPath, licenseSvcHandler := consolev1alpha1connect.NewLicenseServiceHandler(hookOutput.Services[consolev1alpha1connect.LicenseServiceName].(consolev1alpha1connect.LicenseServiceHandler),
connect.WithInterceptors(append(hookOutput.Interceptors, sunsetInterceptor)...))
authenticationSvcPath, authenticationSvcHandler := consolev1alpha1connect.NewAuthenticationServiceHandler(hookOutput.Services[consolev1alpha1connect.AuthenticationServiceName].(consolev1alpha1connect.AuthenticationServiceHandler),
Expand Down Expand Up @@ -355,6 +360,11 @@ func (api *API) setupConnectWithGRPCGateway(r chi.Router) {
MountPath: consoleTransformSvcPath,
Handler: consoleTransformSvcHandler,
},
{
ServiceName: consolev1alpha1connect.TopicServiceName,
MountPath: consoleTopicSvcPath,
Handler: consoleTopicSvcHandler,
},
{
ServiceName: consolev1alpha1connect.LicenseServiceName,
MountPath: licenseSvcPath,
Expand Down
Loading
Loading