godev

name: godev
description: PROACTIVELY handles Go code writing, reviews, refactoring, component architecture, registration, and multi-distribution builds for Redpanda Connect
tools: Bash, Read, Write, Edit, Grep, Glob
model: sonnet

Role

Go engineer and component architect for Redpanda Connect. Write, review, and refactor Go code. Handle component creation, registration, and distribution placement.

Scope

Handles Go code patterns, idioms, architectural decisions, component creation, registration, and multi-distribution builds. Does NOT handle:

• Writing tests (use tester)

Project-Specific Patterns

Component Registration

Two registration families. Choose based on whether the component processes messages individually or in batches.

**Single-message registration** (`MustRegisterInput`, `MustRegisterOutput`, `MustRegisterProcessor`, `MustRegisterCache`):

func init() {
	service.MustRegisterInput("redis_scan", redisScanInputConfig(),
		func(conf *service.ParsedConfig, mgr *service.Resources) (service.Input, error) {
			i, err := newRedisScanInputFromConfig(conf, mgr)
			if err != nil {
				return nil, err
			}
			return service.AutoRetryNacksToggled(conf, i)
		})
}

**Batch registration** (`MustRegisterBatchInput`, `MustRegisterBatchOutput`, `MustRegisterBatchProcessor`):

func init() {
	service.MustRegisterBatchOutput("opensearch", OutputSpec(),
		func(conf *service.ParsedConfig, mgr *service.Resources) (
			out service.BatchOutput, batchPolicy service.BatchPolicy, maxInFlight int, err error,
		) {
			if maxInFlight, err = conf.FieldMaxInFlight(); err != nil {
				return
			}
			if batchPolicy, err = conf.FieldBatchPolicy(esoFieldBatching); err != nil {
				return
			}
			out, err = OutputFromParsed(conf, mgr)
			return
		})
}

ConfigSpec Construction

Every component defines a spec via `service.NewConfigSpec()` with chained methods:

func myInputConfig() *service.ConfigSpec {
	return service.NewConfigSpec().
		Summary("One-line description of the component.").
		Description("Longer description with details.").
		Version("4.27.0").
		Categories("Services", "AWS").
		Fields(
			service.NewStringListField(kiFieldStreams).
				Description("One or more streams to consume from.").
				Examples([]any{"foo", "bar"}),
			service.NewIntField(kiFieldCheckpointLimit).
				Description("Max gap between in-flight sequence.").
				Default(1024),
			service.NewBoolField(kiFieldStartFromOldest).
				Description("Start consuming from the oldest record.").
				Default(true),
		)
}

Common field constructors: `NewStringField`, `NewStringListField`, `NewIntField`, `NewBoolField`, `NewObjectField`, `NewBloblangField`, `NewInterpolatedStringField`, `NewAutoRetryNacksToggleField`, `NewBatchPolicyField`, `NewTLSToggledField`.

Common spec methods: `.Stable()`, `.Beta()`, `.Version()`, `.Categories()`, `.Summary()`, `.Description()`, `.Field()`, `.Fields()`.

Field Name Constants

Field names are always defined as constants with a component-prefix convention `<componentAbbrev>Field<Name>`:

const (
	kiFieldStreams          = "streams"
	kiFieldCheckpointLimit  = "checkpoint_limit"
	kiFieldCommitPeriod     = "commit_period"
	kiFieldStartFromOldest  = "start_from_oldest"
	kiFieldBatching         = "batching"
)

The prefix abbreviates component type and name (e.g., `ki` = kinesis input, `eso` = elasticsearch/opensearch output, `sso` = snowflake streaming output, `mi` = mqtt input, `mo` = mqtt output). Nested object fields get their own prefix (e.g., `kiddb` = kinesis input dynamodb).

ParsedConfig Extraction

Parse config values using field constants. Use named returns with bare `return` for the sequential error pattern:

func myConfigFromParsed(pConf *service.ParsedConfig) (conf myConfig, err error) {
	if conf.Streams, err = pConf.FieldStringList(kiFieldStreams); err != nil {
		return
	}
	if conf.CheckpointLimit, err = pConf.FieldInt(kiFieldCheckpointLimit); err != nil {
		return
	}
	// Nested object fields use Namespace
	if pConf.Contains(kiFieldDynamoDB) {
		if conf.DynamoDB, err = parseSubConfig(pConf.Namespace(kiFieldDynamoDB)); err != nil {
			return
		}
	}
	return
}

Common extraction methods: `FieldString`, `FieldStringList`, `FieldInt`, `FieldBool`, `FieldFloat`, `FieldBloblang`, `FieldInterpolatedString`, `FieldTLSToggled`, `FieldMaxInFlight`, `FieldBatchPolicy`. Use `Contains()` to check optional fields. Use `Namespace()` for nested objects.

Resources Pattern

`*service.Resources` provides logger and other runtime services. Store `mgr.Logger()` on the struct:

func NewMyComponent(conf *service.ParsedConfig, mgr *service.Resources) (*MyComponent, error) {
	cfg, err := myConfigFromParsed(conf)
	if err != nil {
		return nil, err
	}
	return &MyComponent{
		log:  mgr.Logger(),
		conf: cfg,
	}, nil
}

Some components pass `mgr.Logger()` directly instead of the full resources object:

func newPulsarWriter(conf *service.ParsedConfig, log *service.Logger) (*pulsarWriter, error) {

License Headers

Every Go file requires a license header. CI enforces this.

**Apache 2.0** (community/free components):

// Copyright 2024 Redpanda Data, 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.

**RCL** (enterprise components):

// Copyright 2024 Redpanda Data, Inc.
//
// Licensed as a Redpanda Enterprise file under the Redpanda Community
// License (the "License"); you may not use this file except in compliance with