cockroachdb

Documentation

Index

• Constants
• func ConvertParamMapToRawQuery(queryParams map[string]string) string
• func IsWriteOperation(sqlType SQLStatementType) bool
• func RedactSQL(sql string) string
• type Config
  • func (r Config) Initialize(ctx context.Context, tracer trace.Tracer) (sources.Source, error)
  • func (r Config) SourceConfigType() string
• type SQLStatementType
  • func ClassifySQL(sql string) SQLStatementType
• type Source
  • func (s *Source) ApplyQueryLimits(sql string) (string, error)
  • func (s *Source) CanExecuteWrite(sql string) error
  • func (s *Source) CockroachDBPool() *pgxpool.Pool
  • func (s *Source) EmitTelemetry(ctx context.Context, event TelemetryEvent)
  • func (s *Source) ExecuteTxWithRetry(ctx context.Context, fn func(pgx.Tx) error) error
  • func (s *Source) IsReadOnlyMode() bool
  • func (s *Source) PostgresPool() *pgxpool.Pool
  • func (s *Source) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
  • func (s *Source) SourceType() string
  • func (s *Source) ToConfig() sources.SourceConfig
• type StructuredError
  • func (e *StructuredError) Error() string
• type TelemetryEvent

Constants

const (
	ErrCodeUnauthorized         = "CRDB_UNAUTHORIZED"
	ErrCodeReadOnlyViolation    = "CRDB_READONLY_VIOLATION"
	ErrCodeQueryTimeout         = "CRDB_QUERY_TIMEOUT"
	ErrCodeRowLimitExceeded     = "CRDB_ROW_LIMIT_EXCEEDED"
	ErrCodeInvalidSQL           = "CRDB_INVALID_SQL"
	ErrCodeConnectionFailed     = "CRDB_CONNECTION_FAILED"
	ErrCodeWriteModeRequired    = "CRDB_WRITE_MODE_REQUIRED"
	ErrCodeQueryExecutionFailed = "CRDB_QUERY_EXECUTION_FAILED"
)

MCP Error Codes

const SourceType string = "cockroachdb"

Functions

func ConvertParamMapToRawQuery

func ConvertParamMapToRawQuery(queryParams map[string]string) string

func IsWriteOperation

func IsWriteOperation(sqlType SQLStatementType) bool

IsWriteOperation returns true if the SQL statement modifies data

func RedactSQL

func RedactSQL(sql string) string

RedactSQL redacts sensitive values from SQL for telemetry

Types

type Config

type Config struct {
	Name           string            `yaml:"name" validate:"required"`
	Type           string            `yaml:"type" validate:"required"`
	Host           string            `yaml:"host" validate:"required"`
	Port           string            `yaml:"port" validate:"required"`
	User           string            `yaml:"user" validate:"required"`
	Password       string            `yaml:"password"`
	Database       string            `yaml:"database" validate:"required"`
	QueryParams    map[string]string `yaml:"queryParams"`
	MaxRetries     int               `yaml:"maxRetries"`
	RetryBaseDelay string            `yaml:"retryBaseDelay"`

	// MCP Security Features
	ReadOnlyMode    bool `yaml:"readOnlyMode"`    // Default: true (enforced in Initialize)
	EnableWriteMode bool `yaml:"enableWriteMode"` // Explicit opt-in for write operations
	MaxRowLimit     int  `yaml:"maxRowLimit"`     // Default: 1000
	QueryTimeoutSec int  `yaml:"queryTimeoutSec"` // Default: 30

	// Observability
	EnableTelemetry  bool   `yaml:"enableTelemetry"`  // Default: true
	TelemetryVerbose bool   `yaml:"telemetryVerbose"` // Default: false
	ClusterID        string `yaml:"clusterID"`        // Optional cluster identifier for telemetry
}

func (Config) Initialize

func (r Config) Initialize(ctx context.Context, tracer trace.Tracer) (sources.Source, error)

func (Config) SourceConfigType

func (r Config) SourceConfigType() string

type SQLStatementType

type SQLStatementType int

SQLStatementType represents the type of SQL statement

const (
	SQLTypeUnknown SQLStatementType = iota
	SQLTypeSelect
	SQLTypeInsert
	SQLTypeUpdate
	SQLTypeDelete
	SQLTypeDDL // CREATE, ALTER, DROP
	SQLTypeTruncate
	SQLTypeExplain
	SQLTypeShow
	SQLTypeSet
)

func ClassifySQL

func ClassifySQL(sql string) SQLStatementType

ClassifySQL analyzes a SQL statement and returns its type

type Source

type Source struct {
	Config
	Pool *pgxpool.Pool
}

func (*Source) ApplyQueryLimits

func (s *Source) ApplyQueryLimits(sql string) (string, error)

ApplyQueryLimits applies row limits to a SQL query for MCP security compliance. Context timeout management is the responsibility of the caller (following Go best practices). Returns potentially modified SQL with LIMIT clause for SELECT queries.

func (*Source) CanExecuteWrite

func (s *Source) CanExecuteWrite(sql string) error

CanExecuteWrite checks if a write operation is allowed

func (*Source) CockroachDBPool

func (s *Source) CockroachDBPool() *pgxpool.Pool

func (*Source) EmitTelemetry

func (s *Source) EmitTelemetry(ctx context.Context, event TelemetryEvent)

EmitTelemetry logs a telemetry event in structured JSON format

func (*Source) ExecuteTxWithRetry

func (s *Source) ExecuteTxWithRetry(ctx context.Context, fn func(pgx.Tx) error) error

ExecuteTxWithRetry executes a function within a transaction with automatic retry logic using the official CockroachDB retry mechanism from cockroach-go/v2

func (*Source) IsReadOnlyMode

func (s *Source) IsReadOnlyMode() bool

IsReadOnlyMode returns whether the source is in read-only mode

func (*Source) PostgresPool

func (s *Source) PostgresPool() *pgxpool.Pool

func (*Source) Query

func (s *Source) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

Query executes a query using the connection pool with MCP security enforcement. For read-only queries, connection-level retry is sufficient. For write operations requiring transaction retry, use ExecuteTxWithRetry directly. Note: Callers should manage context timeouts as needed.

func (*Source) SourceType

func (s *Source) SourceType() string

func (*Source) ToConfig

func (s *Source) ToConfig() sources.SourceConfig

type StructuredError

type StructuredError struct {
	Code    string         `json:"error_code"`
	Message string         `json:"message"`
	Details map[string]any `json:"details,omitempty"`
}

StructuredError represents an MCP-compliant error with error codes

func (*StructuredError) Error

func (e *StructuredError) Error() string

type TelemetryEvent

type TelemetryEvent struct {
	Timestamp    time.Time         `json:"timestamp"`
	ToolName     string            `json:"tool_name"`
	ClusterID    string            `json:"cluster_id"`
	Database     string            `json:"database"`
	User         string            `json:"user"`
	SQLRedacted  string            `json:"sql_redacted"` // Query with values redacted
	Status       string            `json:"status"`       // "success" | "failure"
	ErrorCode    string            `json:"error_code,omitempty"`
	ErrorMsg     string            `json:"error_msg,omitempty"`
	LatencyMs    int64             `json:"latency_ms"`
	RowsAffected int64             `json:"rows_affected,omitempty"`
	Metadata     map[string]string `json:"metadata,omitempty"`
}

TelemetryEvent represents a structured telemetry event for MCP tool calls