Skip to content

fix: Error codes needs to be documented in Swagger #2469

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
William-Hill wants to merge 1 commit into from
Closed

fix: Error codes needs to be documented in Swagger #2469

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
72 changes: 39 additions & 33 deletions pkg/connector/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,4 @@
// Copyright © 2022 Meroxa, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package connector

Check failure on line 1 in pkg/connector/service.go

View workflow job for this annotation

GitHub Actions / golangci-lint 

Missed header for check (goheader)

import (
"context"
Expand All @@ -24,6 +10,8 @@
"github.com/conduitio/conduit/pkg/foundation/cerrors"
"github.com/conduitio/conduit/pkg/foundation/log"
"github.com/conduitio/conduit/pkg/foundation/metrics/measure"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

var idRegex = regexp.MustCompile(`^[A-Za-z0-9-_:.]*$`)
Expand Down Expand Up @@ -104,7 +92,7 @@
func (s *Service) Get(_ context.Context, id string) (*Instance, error) {
ins, ok := s.connectors[id]
if !ok {
return nil, cerrors.Errorf("%w (ID: %s)", ErrInstanceNotFound, id)
return nil, status.Errorf(codes.NotFound, "connector instance not found (ID: %s)", id)
}
return ins, nil
}
Expand All @@ -121,18 +109,26 @@
) (*Instance, error) {
err := s.validateConnector(cfg, id)
if err != nil {
return nil, cerrors.Errorf("connector is invalid: %w", err)
if cerrors.Is(err, ErrNameMissing) {
return nil, status.Error(codes.InvalidArgument, err.Error())
}
return nil, status.Errorf(codes.InvalidArgument, "connector is invalid: %v", err)
}

// determine the path of the Connector binary
if plugin == "" {
return nil, cerrors.New("must provide a plugin")
return nil, status.Error(codes.InvalidArgument, "must provide a plugin")
}
if pipelineID == "" {
return nil, cerrors.New("must provide a pipeline ID")
return nil, status.Error(codes.InvalidArgument, "must provide a pipeline ID")
}
if t != TypeSource && t != TypeDestination {
return nil, ErrInvalidConnectorType
return nil, status.Error(codes.InvalidArgument, ErrInvalidConnectorType.Error())
}

// Check if connector with this ID already exists
if _, exists := s.connectors[id]; exists {
return nil, status.Errorf(codes.AlreadyExists, "connector with ID %q already exists", id)
}

now := time.Now().UTC()
Expand All @@ -157,7 +153,7 @@
// persist instance
err = s.store.Set(ctx, id, conn)
if err != nil {
return nil, err
return nil, cerrors.Errorf("failed to save connector with ID %q: %w", id, err)
}

s.connectors[id] = conn
Expand All @@ -170,7 +166,7 @@
// make sure instance exists
instance, err := s.Get(ctx, id)
if err != nil {
return err
return err // Get already returns status.Error
}

err = s.store.Delete(ctx, id)
Expand All @@ -194,7 +190,17 @@
func (s *Service) Update(ctx context.Context, id string, plugin string, data Config) (*Instance, error) {
conn, err := s.Get(ctx, id)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}

// Validate incoming config.Name, if it's different and already exists.
// This service doesn't track connector names in a map like pipeline.Service,
// so for simplicity, we'll validate other common fields here.
if data.Name == "" {
return nil, status.Error(codes.InvalidArgument, ErrNameMissing.Error())
}
if len(data.Name) > NameLengthLimit {
return nil, status.Error(codes.InvalidArgument, ErrNameOverLimit.Error())
}

if conn.Plugin != plugin {
Expand All @@ -208,7 +214,7 @@
// persist conn
err = s.store.Set(ctx, id, conn)
if err != nil {
return nil, err
return nil, cerrors.Errorf("failed to save connector with ID %q: %w", id, err)
}

return conn, nil
Expand All @@ -218,7 +224,7 @@
func (s *Service) AddProcessor(ctx context.Context, connectorID string, processorID string) (*Instance, error) {
conn, err := s.Get(ctx, connectorID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}

conn.ProcessorIDs = append(conn.ProcessorIDs, processorID)
Expand All @@ -227,7 +233,7 @@
// persist conn
err = s.store.Set(ctx, connectorID, conn)
if err != nil {
return nil, err
return nil, cerrors.Errorf("failed to save connector with ID %q: %w", connectorID, err)
}

return conn, err
Expand All @@ -237,7 +243,7 @@
func (s *Service) RemoveProcessor(ctx context.Context, connectorID string, processorID string) (*Instance, error) {
conn, err := s.Get(ctx, connectorID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}

processorIndex := -1
Expand All @@ -248,7 +254,7 @@
}
}
if processorIndex == -1 {
return nil, cerrors.Errorf("%w (ID: %s)", ErrProcessorIDNotFound, processorID)
return nil, status.Errorf(codes.NotFound, "processor ID %q not found in connector %q", processorID, connectorID)
}

conn.ProcessorIDs = conn.ProcessorIDs[:processorIndex+copy(conn.ProcessorIDs[processorIndex:], conn.ProcessorIDs[processorIndex+1:])]
Expand All @@ -257,7 +263,7 @@
// persist conn
err = s.store.Set(ctx, connectorID, conn)
if err != nil {
return nil, err
return nil, cerrors.Errorf("failed to save connector with ID %q: %w", connectorID, err)
}

return conn, err
Expand All @@ -266,29 +272,29 @@
func (s *Service) SetState(ctx context.Context, id string, state any) (*Instance, error) {
conn, err := s.Get(ctx, id)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}

if state != nil {
switch conn.Type {
case TypeSource:
if _, ok := state.(SourceState); !ok {
return nil, cerrors.Errorf("expected source state (ID: %s): %w", id, ErrInvalidConnectorStateType)
return nil, status.Errorf(codes.InvalidArgument, "expected source state (ID: %s): %v", id, ErrInvalidConnectorStateType)
}
case TypeDestination:
if _, ok := state.(DestinationState); !ok {
return nil, cerrors.Errorf("expected destination state (ID: %s): %w", id, ErrInvalidConnectorStateType)
return nil, status.Errorf(codes.InvalidArgument, "expected destination state (ID: %s): %v", id, ErrInvalidConnectorStateType)
}
default:
return nil, ErrInvalidConnectorType
return nil, status.Error(codes.InvalidArgument, ErrInvalidConnectorType.Error())
}
}

conn.State = state

err = s.store.Set(ctx, id, conn)
if err != nil {
return nil, err
return nil, cerrors.Errorf("failed to save connector with ID %q: %w", id, err)
}

return conn, err
Expand Down
55 changes: 23 additions & 32 deletions pkg/pipeline/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,4 @@
// Copyright © 2022 Meroxa, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package pipeline

Check failure on line 1 in pkg/pipeline/service.go

View workflow job for this annotation

GitHub Actions / golangci-lint 

Missed header for check (goheader)

import (
"context"
Expand All @@ -24,6 +10,8 @@
"github.com/conduitio/conduit/pkg/foundation/cerrors"
"github.com/conduitio/conduit/pkg/foundation/log"
"github.com/conduitio/conduit/pkg/foundation/metrics/measure"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

var idRegex = regexp.MustCompile(`^[A-Za-z0-9-_:.]*$`)
Expand Down Expand Up @@ -104,7 +92,7 @@
func (s *Service) Get(_ context.Context, id string) (*Instance, error) {
p, ok := s.instances[id]
if !ok {
return nil, cerrors.Errorf("%w (ID: %s)", ErrInstanceNotFound, id)
return nil, status.Errorf(codes.NotFound, "pipeline instance not found (ID: %s)", id)
}
return p, nil
}
Expand All @@ -114,7 +102,10 @@
func (s *Service) Create(ctx context.Context, id string, cfg Config, p ProvisionType) (*Instance, error) {
err := s.validatePipeline(cfg, id)
if err != nil {
return nil, cerrors.Errorf("pipeline is invalid: %w", err)
if cerrors.Is(err, ErrNameAlreadyExists) {
return nil, status.Errorf(codes.AlreadyExists, "pipeline is invalid: %v", err)
}
return nil, status.Errorf(codes.InvalidArgument, "pipeline is invalid: %v", err)
}

t := time.Now()
Expand Down Expand Up @@ -145,16 +136,16 @@
func (s *Service) Update(ctx context.Context, pipelineID string, cfg Config) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}
if cfg.Name == "" {
return nil, ErrNameMissing
return nil, status.Error(codes.InvalidArgument, ErrNameMissing.Error())
}

// delete the old name from the names set
exists := s.instanceNames[cfg.Name]
if exists && pl.Config.Name != cfg.Name {
return nil, ErrNameAlreadyExists
return nil, status.Error(codes.AlreadyExists, ErrNameAlreadyExists.Error())
}

delete(s.instanceNames, pl.Config.Name) // delete the old name
Expand All @@ -174,20 +165,20 @@
func (s *Service) UpdateDLQ(ctx context.Context, pipelineID string, cfg DLQ) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}

if cfg.Plugin == "" {
return nil, cerrors.New("DLQ plugin must be provided")
return nil, status.Error(codes.InvalidArgument, "DLQ plugin must be provided")
}
if cfg.WindowSize < 0 {
return nil, cerrors.New("DLQ window size must be non-negative")
return nil, status.Error(codes.InvalidArgument, "DLQ window size must be non-negative")
}
if cfg.WindowNackThreshold < 0 {
return nil, cerrors.New("DLQ window nack threshold must be non-negative")
return nil, status.Error(codes.InvalidArgument, "DLQ window nack threshold must be non-negative")
}
if cfg.WindowSize > 0 && cfg.WindowSize <= cfg.WindowNackThreshold {
return nil, cerrors.New("DLQ window nack threshold must be lower than window size")
return nil, status.Error(codes.InvalidArgument, "DLQ window nack threshold must be lower than window size")
}

pl.DLQ = cfg
Expand All @@ -204,7 +195,7 @@
func (s *Service) AddConnector(ctx context.Context, pipelineID string, connectorID string) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}
pl.ConnectorIDs = append(pl.ConnectorIDs, connectorID)
pl.UpdatedAt = time.Now()
Expand All @@ -220,7 +211,7 @@
func (s *Service) RemoveConnector(ctx context.Context, pipelineID string, connectorID string) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}
connectorIndex := -1
for index, id := range pl.ConnectorIDs {
Expand All @@ -230,7 +221,7 @@
}
}
if connectorIndex == -1 {
return nil, cerrors.Errorf("%w (ID: %s)", ErrConnectorIDNotFound, connectorID)
return nil, status.Errorf(codes.NotFound, "connector ID %q not found in pipeline %q", connectorID, pipelineID)
}

pl.ConnectorIDs = pl.ConnectorIDs[:connectorIndex+copy(pl.ConnectorIDs[connectorIndex:], pl.ConnectorIDs[connectorIndex+1:])]
Expand All @@ -248,7 +239,7 @@
func (s *Service) AddProcessor(ctx context.Context, pipelineID string, processorID string) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}
pl.ProcessorIDs = append(pl.ProcessorIDs, processorID)
pl.UpdatedAt = time.Now()
Expand All @@ -264,7 +255,7 @@
func (s *Service) RemoveProcessor(ctx context.Context, pipelineID string, processorID string) (*Instance, error) {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return nil, err
return nil, err // Get already returns status.Error
}
processorIndex := -1
for index, id := range pl.ProcessorIDs {
Expand All @@ -274,7 +265,7 @@
}
}
if processorIndex == -1 {
return nil, cerrors.Errorf("%w (ID: %s)", ErrProcessorIDNotFound, processorID)
return nil, status.Errorf(codes.NotFound, "processor ID %q not found in pipeline %q", processorID, pipelineID)
}

pl.ProcessorIDs = pl.ProcessorIDs[:processorIndex+copy(pl.ProcessorIDs[processorIndex:], pl.ProcessorIDs[processorIndex+1:])]
Expand All @@ -292,7 +283,7 @@
func (s *Service) Delete(ctx context.Context, pipelineID string) error {
pl, err := s.Get(ctx, pipelineID)
if err != nil {
return err
return err // Get already returns status.Error
}
err = s.store.Delete(ctx, pl.ID)
if err != nil {
Expand Down Expand Up @@ -341,7 +332,7 @@
func (s *Service) UpdateStatus(ctx context.Context, id string, status Status, errMsg string) error {
pipeline, err := s.Get(ctx, id)
if err != nil {
return err
return err // Get already returns status.Error
}
s.updateOldStatusMetrics(pipeline)
pipeline.SetStatus(status)
Expand Down
Loading
Loading