[Aider] Phase 0

This commit is contained in:
Tanishq Dubey 2025-05-10 18:18:58 -04:00
parent 2f0debf608
commit 1ae06781d6
No known key found for this signature in database
GPG Key ID: CFC1931B84DFC3F9
14 changed files with 4438 additions and 55 deletions

33
Makefile Normal file
View File

@ -0,0 +1,33 @@
# File: Makefile
.PHONY: all generate clean test
# Variables
GOLANGCI_LINT_VERSION := v1.55.2 # Or your preferred version
all: generate test
generate:
@echo "Generating Go code from Protobuf definitions..."
@./scripts/gen-proto.sh
# Placeholder for future commands
clean:
@echo "Cleaning up generated files and build artifacts..."
@rm -f ./api/v1alpha1/*.pb.go
@rm -f kat-agent katcall
test:
@echo "Running tests..."
@go test ./...
lint:
@echo "Running linter..."
@if ! command -v golangci-lint &> /dev/null; then \
echo "golangci-lint not found. Installing..."; \
go install github.com/golangci/golangci-lint/cmd/golangci-lint@$(GOLANGCI_LINT_VERSION); \
fi
@golangci-lint run
# Add to go.mod if not already present by go install
# go get google.golang.org/protobuf/cmd/protoc-gen-go
# go get google.golang.org/grpc/cmd/protoc-gen-go-grpc (if you plan to use gRPC services soon)

3458
api/v1alpha1/kat.pb.go Normal file

File diff suppressed because it is too large Load Diff

345
api/v1alpha1/kat.proto Normal file
View File

@ -0,0 +1,345 @@
// File: api/v1alpha1/kat.proto
syntax = "proto3";
package v1alpha1;
option go_package = "git.dws.rip/dubey/kat"; // Adjust to your actual go module path
import "google/protobuf/timestamp.proto";
// Common Metadata (RFC 3.2, Phase 0 Docs)
message ObjectMeta {
string name = 1;
string namespace = 2;
string uid = 3;
int64 generation = 4;
string resource_version = 5; // e.g., etcd ModRevision
google.protobuf.Timestamp creation_timestamp = 6;
map<string, string> labels = 7;
map<string, string> annotations = 8;
}
// Workload (RFC 3.2)
enum WorkloadType {
WORKLOAD_TYPE_UNSPECIFIED = 0;
SERVICE = 1;
JOB = 2;
DAEMON_SERVICE = 3;
}
message GitSource {
string repository = 1;
string branch = 2;
string tag = 3;
string commit = 4;
}
message WorkloadSource {
oneof source_type {
string image = 1; // Direct image reference
GitSource git = 2; // Build from Git
}
string cache_image = 3; // Optional: Registry path for build cache layers (used with git source)
}
enum UpdateStrategyType {
UPDATE_STRATEGY_TYPE_UNSPECIFIED = 0;
ROLLING = 1;
SIMULTANEOUS = 2;
}
message RollingUpdateStrategy {
string max_surge = 1; // Can be int or percentage string e.g., "1" or "10%"
}
message UpdateStrategy {
UpdateStrategyType type = 1;
RollingUpdateStrategy rolling = 2; // Relevant if type is ROLLING
}
enum RestartCondition {
RESTART_CONDITION_UNSPECIFIED = 0;
NEVER = 1;
MAX_COUNT = 2;
ALWAYS = 3;
}
message RestartPolicy {
RestartCondition condition = 1;
int32 max_restarts = 2; // Used if condition=MAX_COUNT
int32 reset_seconds = 3; // Used if condition=MAX_COUNT
}
message Toleration {
string key = 1;
enum Operator {
OPERATOR_UNSPECIFIED = 0;
EXISTS = 1;
EQUAL = 2;
}
Operator operator = 2;
string value = 3; // Needed if operator=EQUAL
enum Effect {
EFFECT_UNSPECIFIED = 0;
NO_SCHEDULE = 1;
PREFER_NO_SCHEDULE = 2;
// NO_EXECUTE (not in RFC v1 scope for tolerations, but common)
}
Effect effect = 4;
}
message EnvVar {
string name = 1;
string value = 2;
}
message VolumeMount {
string name = 1; // Volume name from spec.volumes
string mount_path = 2; // Path inside container
string sub_path = 3; // Optional: Mount sub-directory
bool read_only = 4; // Optional: Default false
}
message ResourceRequests {
string cpu = 1; // e.g., "100m"
string memory = 2; // e.g., "64Mi"
}
message ResourceLimits {
string cpu = 1; // e.g., "1"
string memory = 2; // e.g., "256Mi"
}
enum GPUDriver {
GPU_DRIVER_UNSPECIFIED = 0;
ANY = 1;
NVIDIA = 2;
AMD = 3;
}
message GPUSpec {
GPUDriver driver = 1;
int32 min_vram_mb = 2; // Minimum GPU memory required
}
message ContainerResources {
ResourceRequests requests = 1;
ResourceLimits limits = 2;
GPUSpec gpu = 3;
}
message Container {
string name = 1; // Optional: Informational name
repeated string command = 2;
repeated string args = 3;
repeated EnvVar env = 4;
repeated VolumeMount volume_mounts = 5;
ContainerResources resources = 6;
}
message SimpleClusterStorageVolumeSource {
// Empty, implies agent creates dir under volumeBasePath
}
enum HostPathType {
HOST_PATH_TYPE_UNSPECIFIED = 0; // No check, mount whatever is there or fail
DIRECTORY_OR_CREATE = 1;
DIRECTORY = 2;
FILE_OR_CREATE = 3;
FILE = 4;
SOCKET = 5;
}
message HostMountVolumeSource {
string host_path = 1; // Absolute path on host
HostPathType ensure_type = 2; // Optional: Type to ensure/check
}
message Volume {
string name = 1; // Name referenced by volumeMounts
oneof volume_source {
SimpleClusterStorageVolumeSource simple_cluster_storage = 2;
HostMountVolumeSource host_mount = 3;
}
}
message WorkloadSpec {
WorkloadType type = 1;
WorkloadSource source = 2;
int32 replicas = 3; // Required for SERVICE
UpdateStrategy update_strategy = 4;
RestartPolicy restart_policy = 5;
map<string, string> node_selector = 6;
repeated Toleration tolerations = 7;
Container container = 8;
repeated Volume volumes = 9;
}
message WorkloadStatus {
// Placeholder for Phase 0. Will be expanded later.
// Example fields:
// int32 observed_generation = 1;
// int32 ready_replicas = 2;
// string condition = 3; // e.g., "Available", "Progressing", "Failed"
}
message Workload {
ObjectMeta metadata = 1;
WorkloadSpec spec = 2;
WorkloadStatus status = 3;
}
// VirtualLoadBalancer (RFC 3.3)
message PortSpec {
string name = 1; // Optional: e.g., "web", "grpc"
int32 container_port = 2; // Port app listens on in container
string protocol = 3; // Optional: TCP | UDP. Default TCP.
}
message ExecHealthCheck {
repeated string command = 1; // Exit 0 = healthy
}
message HealthCheck {
ExecHealthCheck exec = 1;
int32 initial_delay_seconds = 2;
int32 period_seconds = 3;
int32 timeout_seconds = 4;
int32 success_threshold = 5;
int32 failure_threshold = 6;
}
message IngressRule {
string host = 1;
string path = 2;
string service_port_name = 3; // Name of port from PortSpec
int32 service_port = 4; // Port number from PortSpec (overrides name)
bool tls = 5; // Signal for ACME
}
message VirtualLoadBalancerSpec {
repeated PortSpec ports = 1;
HealthCheck health_check = 2;
repeated IngressRule ingress = 3;
}
message VirtualLoadBalancer {
ObjectMeta metadata = 1; // Name likely matches Workload name
VirtualLoadBalancerSpec spec = 2;
// VirtualLoadBalancerStatus status = 3; // Placeholder
}
// JobDefinition (RFC 3.4)
message JobDefinitionSpec {
string schedule = 1; // Optional: Cron schedule
int32 completions = 2; // Optional: Default 1
int32 parallelism = 3; // Optional: Default 1
int32 active_deadline_seconds = 4; // Optional
int32 backoff_limit = 5; // Optional: Default 3
}
message JobDefinition {
ObjectMeta metadata = 1; // Name likely matches Workload name
JobDefinitionSpec spec = 2;
// JobDefinitionStatus status = 3; // Placeholder
}
// BuildDefinition (RFC 3.5)
message BuildCache {
string registry_path = 1; // e.g., "myreg.com/cache/myapp"
}
message BuildDefinitionSpec {
string build_context = 1; // Optional: Path relative to repo root. Defaults to "."
string dockerfile_path = 2; // Optional: Path relative to buildContext. Defaults to "./Dockerfile"
map<string, string> build_args = 3; // Optional
string target_stage = 4; // Optional
string platform = 5; // Optional: e.g., "linux/arm64"
BuildCache cache = 6; // Optional
}
message BuildDefinition {
ObjectMeta metadata = 1; // Name likely matches Workload name
BuildDefinitionSpec spec = 2;
// BuildDefinitionStatus status = 3; // Placeholder
}
// Namespace (RFC 3.7)
message NamespaceSpec {
// Potentially finalizers or other future spec fields
}
message NamespaceStatus {
// string phase = 1; // e.g., "Active", "Terminating"
}
message Namespace {
ObjectMeta metadata = 1;
NamespaceSpec spec = 2;
NamespaceStatus status = 3;
}
// Node (Internal Representation - RFC 3.8)
message NodeResources {
string cpu = 1; // e.g., "2000m"
string memory = 2; // e.g., "4096Mi"
map<string, string> custom_resources = 3; // e.g., for GPUs "nvidia.com/gpu: 2"
}
message NodeStatusDetails {
NodeResources capacity = 1;
NodeResources allocatable = 2;
// repeated WorkloadInstanceStatus workload_instances = 3; // Define later
// OverlayNetworkStatus overlay_network = 4; // Define later
string condition = 5; // e.g., "Ready", "NotReady", "Draining"
google.protobuf.Timestamp last_heartbeat_time = 6;
}
message Taint { // From RFC 1.5, used in NodeSpec
string key = 1;
string value = 2;
enum Effect {
EFFECT_UNSPECIFIED = 0;
NO_SCHEDULE = 1;
PREFER_NO_SCHEDULE = 2;
// NO_EXECUTE (not in RFC v1 scope for taints, but common)
}
Effect effect = 3;
}
message NodeSpec {
repeated Taint taints = 1;
string overlay_subnet = 2; // Assigned by leader
// string provider_id = 3; // Cloud provider specific ID
// map<string, string> labels = 4; // Node labels, distinct from metadata.labels
// map<string, string> annotations = 5; // Node annotations
}
message Node {
ObjectMeta metadata = 1; // Name is the unique node name/ID
NodeSpec spec = 2;
NodeStatusDetails status = 3;
}
// ClusterConfiguration (RFC 3.9)
message ClusterConfigurationSpec {
string cluster_cidr = 1;
string service_cidr = 2;
int32 node_subnet_bits = 3;
string cluster_domain = 4;
int32 agent_port = 5;
int32 api_port = 6;
int32 etcd_peer_port = 7;
int32 etcd_client_port = 8;
string volume_base_path = 9;
string backup_path = 10;
int32 backup_interval_minutes = 11;
int32 agent_tick_seconds = 12;
int32 node_loss_timeout_seconds = 13;
}
message ClusterConfiguration {
ObjectMeta metadata = 1; // e.g., name of the cluster
ClusterConfigurationSpec spec = 2;
// ClusterConfigurationStatus status = 3; // Placeholder
}

18
examples/cluster.kat Normal file
View File

@ -0,0 +1,18 @@
apiVersion: kat.dws.rip/v1alpha1
kind: ClusterConfiguration
metadata:
name: my-kat-cluster
spec:
clusterCIDR: "10.100.0.0/16"
serviceCIDR: "10.200.0.0/16"
nodeSubnetBits: 7 # Results in /23 node subnets (e.g., 10.100.0.0/23, 10.100.2.0/23)
clusterDomain: "kat.example.local" # Overriding default
apiPort: 9115
agentPort: 9116
etcdPeerPort: 2380
etcdClientPort: 2379
volumeBasePath: "/opt/kat/volumes" # Overriding default
backupPath: "/opt/kat/backups" # Overriding default
backupIntervalMinutes: 60
agentTickSeconds: 10
nodeLossTimeoutSeconds: 45

View File

@ -0,0 +1,15 @@
apiVersion: kat.dws.rip/v1alpha1
kind: VirtualLoadBalancer
metadata:
name: my-simple-nginx # Should match workload name
namespace: default
spec:
ports:
- name: http
containerPort: 80
protocol: TCP
healthCheck:
exec:
command: ["curl", "-f", "http://localhost/"] # Nginx doesn't have curl by default, this is illustrative
initialDelaySeconds: 5
periodSeconds: 10

View File

@ -0,0 +1,21 @@
apiVersion: kat.dws.rip/v1alpha1
kind: Workload
metadata:
name: my-simple-nginx
namespace: default
spec:
type: SERVICE
source:
image: "nginx:latest"
replicas: 2
restartPolicy:
condition: ALWAYS
container:
name: nginx-container
resources:
requests:
cpu: "50m"
memory: "64Mi"
limits:
cpu: "100m"
memory: "128Mi"

6
go.mod
View File

@ -1,8 +1,10 @@
module git.dws.rip/dubey/kat
go 1.21
go 1.22
toolchain go1.24.2
require (
google.golang.org/protobuf v1.31.0
google.golang.org/protobuf v1.36.6
gopkg.in/yaml.v3 v3.0.1
)

10
go.sum Normal file
View File

@ -0,0 +1,10 @@
github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY=
google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

323
internal/config/parse.go Normal file
View File

@ -0,0 +1,323 @@
// File: internal/config/parse.go
package config
import (
"fmt"
"io/ioutil"
"net"
"os"
pb "git.dws.rip/dubey/kat/api/v1alpha1" // Adjust to your actual go module path
"gopkg.in/yaml.v3" // Add to go.mod: go get gopkg.in/yaml.v3
)
var _ = yaml.Unmarshal // Used for Quadlet parsing
// ParseClusterConfiguration reads, unmarshals, and validates a cluster.kat file.
func ParseClusterConfiguration(filePath string) (*pb.ClusterConfiguration, error) {
if _, err := os.Stat(filePath); os.IsNotExist(err) {
return nil, fmt.Errorf("cluster configuration file not found: %s", filePath)
}
yamlFile, err := ioutil.ReadFile(filePath)
if err != nil {
return nil, fmt.Errorf("failed to read cluster configuration file %s: %w", filePath, err)
}
var config pb.ClusterConfiguration
// We expect the YAML to have top-level keys like 'apiVersion', 'kind', 'metadata', 'spec'
// but our proto is just the ClusterConfiguration message.
// So, we'll unmarshal into a temporary map to extract the 'spec' and 'metadata'.
var rawConfigMap map[string]interface{}
if err = yaml.Unmarshal(yamlFile, &rawConfigMap); err != nil {
return nil, fmt.Errorf("failed to unmarshal YAML from %s: %w", filePath, err)
}
// Quick check for kind
kind, ok := rawConfigMap["kind"].(string)
if !ok || kind != "ClusterConfiguration" {
return nil, fmt.Errorf("invalid kind in %s: expected ClusterConfiguration, got %v", filePath, rawConfigMap["kind"])
}
metadataMap, ok := rawConfigMap["metadata"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("metadata section not found or invalid in %s", filePath)
}
metadataBytes, err := yaml.Marshal(metadataMap)
if err != nil {
return nil, fmt.Errorf("failed to re-marshal metadata: %w", err)
}
if err = yaml.Unmarshal(metadataBytes, &config.Metadata); err != nil {
return nil, fmt.Errorf("failed to unmarshal metadata into proto: %w", err)
}
specMap, ok := rawConfigMap["spec"].(map[string]interface{})
if !ok {
return nil, fmt.Errorf("spec section not found or invalid in %s", filePath)
}
specBytes, err := yaml.Marshal(specMap)
if err != nil {
return nil, fmt.Errorf("failed to re-marshal spec: %w", err)
}
if err = yaml.Unmarshal(specBytes, &config.Spec); err != nil {
return nil, fmt.Errorf("failed to unmarshal spec into proto: %w", err)
}
SetClusterConfigDefaults(&config)
if err := ValidateClusterConfiguration(&config); err != nil {
return nil, fmt.Errorf("invalid cluster configuration in %s: %w", filePath, err)
}
return &config, nil
}
// SetClusterConfigDefaults applies default values to the ClusterConfiguration spec.
func SetClusterConfigDefaults(config *pb.ClusterConfiguration) {
if config.Spec == nil {
config.Spec = &pb.ClusterConfigurationSpec{}
}
s := config.Spec
if s.ClusterDomain == "" {
s.ClusterDomain = DefaultClusterDomain
}
if s.AgentPort == 0 {
s.AgentPort = DefaultAgentPort
}
if s.ApiPort == 0 {
s.ApiPort = DefaultApiPort
}
if s.EtcdPeerPort == 0 {
s.EtcdPeerPort = DefaultEtcdPeerPort
}
if s.EtcdClientPort == 0 {
s.EtcdClientPort = DefaultEtcdClientPort
}
if s.VolumeBasePath == "" {
s.VolumeBasePath = DefaultVolumeBasePath
}
if s.BackupPath == "" {
s.BackupPath = DefaultBackupPath
}
if s.BackupIntervalMinutes == 0 {
s.BackupIntervalMinutes = DefaultBackupIntervalMins
}
if s.AgentTickSeconds == 0 {
s.AgentTickSeconds = DefaultAgentTickSeconds
}
if s.NodeLossTimeoutSeconds == 0 {
s.NodeLossTimeoutSeconds = DefaultNodeLossTimeoutSec
if s.AgentTickSeconds > 0 { // If agent tick is set, derive from it
s.NodeLossTimeoutSeconds = s.AgentTickSeconds * 4 // Example: 4 ticks
}
}
if s.NodeSubnetBits == 0 {
s.NodeSubnetBits = DefaultNodeSubnetBits
}
}
// ValidateClusterConfiguration performs basic validation on the ClusterConfiguration.
func ValidateClusterConfiguration(config *pb.ClusterConfiguration) error {
if config.Metadata == nil || config.Metadata.Name == "" {
return fmt.Errorf("metadata.name is required")
}
if config.Spec == nil {
return fmt.Errorf("spec is required")
}
s := config.Spec
if s.ClusterCidr == "" {
return fmt.Errorf("spec.clusterCIDR is required")
}
if _, _, err := net.ParseCIDR(s.ClusterCidr); err != nil {
return fmt.Errorf("invalid spec.clusterCIDR %s: %w", s.ClusterCidr, err)
}
if s.ServiceCidr == "" {
return fmt.Errorf("spec.serviceCIDR is required")
}
if _, _, err := net.ParseCIDR(s.ServiceCidr); err != nil {
return fmt.Errorf("invalid spec.serviceCIDR %s: %w", s.ServiceCidr, err)
}
// Validate ports
ports := []struct {
name string
port int32
}{
{"agentPort", s.AgentPort}, {"apiPort", s.ApiPort},
{"etcdPeerPort", s.EtcdPeerPort}, {"etcdClientPort", s.EtcdClientPort},
}
for _, p := range ports {
if p.port <= 0 || p.port > 65535 {
return fmt.Errorf("invalid port for %s: %d. Must be between 1 and 65535", p.name, p.port)
}
}
// Check for port conflicts among configured ports
portSet := make(map[int32]string)
for _, p := range ports {
if existing, found := portSet[p.port]; found {
return fmt.Errorf("port conflict: %s (%d) and %s (%d) use the same port", p.name, p.port, existing, p.port)
}
portSet[p.port] = p.name
}
if s.NodeSubnetBits <= 0 || s.NodeSubnetBits >= 32 {
return fmt.Errorf("invalid spec.nodeSubnetBits: %d. Must be > 0 and < 32", s.NodeSubnetBits)
}
// Validate nodeSubnetBits against clusterCIDR prefix length
_, clusterNet, _ := net.ParseCIDR(s.ClusterCidr)
clusterPrefixLen, _ := clusterNet.Mask.Size()
if int(s.NodeSubnetBits) <= clusterPrefixLen {
// This logic might be too simple. NodeSubnetBits is the number of *additional* bits for the subnet *within* the cluster prefix.
// So, the resulting node subnet prefix length would be clusterPrefixLen + s.NodeSubnetBits.
// This must be less than 32 (or 31 for usable IPs).
// The RFC states: "Default 7 (yielding /23 subnets if clusterCIDR=/16)"
// So if clusterCIDR is /16, node subnet is / (16+7) = /23. This is valid.
// A node subnet prefix length must be > clusterPrefixLen and < 32.
if (clusterPrefixLen + int(s.NodeSubnetBits)) >= 32 {
return fmt.Errorf("spec.nodeSubnetBits (%d) combined with clusterCIDR prefix (%d) results in an invalid subnet size (>= /32)", s.NodeSubnetBits, clusterPrefixLen)
}
} else {
// This case seems unlikely if nodeSubnetBits is the number of bits for the node part.
// Let's assume nodeSubnetBits is the number of bits *after* the cluster prefix that define the node subnet.
// e.g. cluster 10.0.0.0/8, nodeSubnetBits=8 -> node subnets are /16.
}
if s.BackupIntervalMinutes < 0 { // 0 could mean disabled, but RFC implies positive
return fmt.Errorf("spec.backupIntervalMinutes must be non-negative")
}
if s.AgentTickSeconds <= 0 {
return fmt.Errorf("spec.agentTickSeconds must be positive")
}
if s.NodeLossTimeoutSeconds <= 0 {
return fmt.Errorf("spec.nodeLossTimeoutSeconds must be positive")
}
if s.NodeLossTimeoutSeconds < s.AgentTickSeconds {
return fmt.Errorf("spec.nodeLossTimeoutSeconds must be greater than or equal to spec.agentTickSeconds")
}
// volumeBasePath and backupPath should be absolute paths, but validation can be tricky
// For now, just check if they are non-empty if specified, defaults handle empty.
// A more robust check would be filepath.IsAbs()
return nil
}
// ParsedQuadletFiles holds the structured data from a Quadlet directory.
type ParsedQuadletFiles struct {
Workload *pb.Workload
VirtualLoadBalancer *pb.VirtualLoadBalancer
JobDefinition *pb.JobDefinition
BuildDefinition *pb.BuildDefinition
// Namespace is typically a cluster-level resource, not part of a workload quadlet bundle.
// If it were, it would be: Namespace *pb.Namespace
// Store raw file contents for potential future use (e.g. annotations, original source)
RawFiles map[string][]byte
}
// ParseQuadletFile unmarshals a single Quadlet file content based on its kind.
// It returns the specific protobuf message.
func ParseQuadletFile(fileName string, content []byte) (interface{}, error) {
var base struct {
ApiVersion string `yaml:"apiVersion"`
Kind string `yaml:"kind"`
}
if err := yaml.Unmarshal(content, &base); err != nil {
return nil, fmt.Errorf("failed to unmarshal base YAML from %s to determine kind: %w", fileName, err)
}
// TODO: Check apiVersion, e.g., base.ApiVersion == "kat.dws.rip/v1alpha1"
var resource interface{}
var err error
switch base.Kind {
case "Workload":
var wl pb.Workload
// Similar to ClusterConfiguration, need to unmarshal metadata and spec separately
// from a temporary map if the proto doesn't match the full YAML structure directly.
// For simplicity in Phase 0, assuming direct unmarshal works if YAML matches proto structure.
// If YAML has apiVersion/kind/metadata/spec at top level, then:
var raw map[string]interface{}
if err = yaml.Unmarshal(content, &raw); err == nil {
if meta, ok := raw["metadata"]; ok {
metaBytes, _ := yaml.Marshal(meta)
yaml.Unmarshal(metaBytes, &wl.Metadata)
}
if spec, ok := raw["spec"]; ok {
specBytes, _ := yaml.Marshal(spec)
yaml.Unmarshal(specBytes, &wl.Spec)
}
}
resource = &wl
case "VirtualLoadBalancer":
var vlb pb.VirtualLoadBalancer
var raw map[string]interface{}
if err = yaml.Unmarshal(content, &raw); err == nil {
if meta, ok := raw["metadata"]; ok {
metaBytes, _ := yaml.Marshal(meta)
yaml.Unmarshal(metaBytes, &vlb.Metadata)
}
if spec, ok := raw["spec"]; ok {
specBytes, _ := yaml.Marshal(spec)
yaml.Unmarshal(specBytes, &vlb.Spec)
}
}
resource = &vlb
// Add cases for JobDefinition, BuildDefinition as they are defined
case "JobDefinition":
var jd pb.JobDefinition
// ... unmarshal logic ...
resource = &jd
case "BuildDefinition":
var bd pb.BuildDefinition
// ... unmarshal logic ...
resource = &bd
default:
return nil, fmt.Errorf("unknown Kind '%s' in file %s", base.Kind, fileName)
}
if err != nil {
return nil, fmt.Errorf("failed to unmarshal YAML for Kind '%s' from %s: %w", base.Kind, fileName, err)
}
// TODO: Add basic validation for each parsed type (e.g., required fields like metadata.name)
return resource, nil
}
// ParseQuadletDirectory processes a map of file contents (from UntarQuadlets).
func ParseQuadletDirectory(files map[string][]byte) (*ParsedQuadletFiles, error) {
parsed := &ParsedQuadletFiles{RawFiles: files}
for fileName, content := range files {
obj, err := ParseQuadletFile(fileName, content)
if err != nil {
return nil, fmt.Errorf("error parsing quadlet file %s: %w", fileName, err)
}
switch v := obj.(type) {
case *pb.Workload:
if parsed.Workload != nil {
return nil, fmt.Errorf("multiple Workload definitions found")
}
parsed.Workload = v
case *pb.VirtualLoadBalancer:
if parsed.VirtualLoadBalancer != nil {
return nil, fmt.Errorf("multiple VirtualLoadBalancer definitions found")
}
parsed.VirtualLoadBalancer = v
// Add cases for JobDefinition, BuildDefinition
}
}
// TODO: Perform cross-Quadlet file validation (e.g., workload.kat must exist)
if parsed.Workload == nil {
return nil, fmt.Errorf("required Workload definition (workload.kat) not found in Quadlet bundle")
}
if parsed.Workload.Metadata == nil || parsed.Workload.Metadata.Name == "" {
return nil, fmt.Errorf("workload.kat must have metadata.name defined")
}
return parsed, nil
}

View File

@ -1,19 +1,16 @@
package config
import (
"io/ioutil"
"os"
"path/filepath"
"strings"
"testing"
"git.dws.rip/dubey/kat/api/v1alpha1"
"google.golang.org/protobuf/types/known/timestamppb"
pb "git.dws.rip/dubey/kat/api/v1alpha1"
)
func createTestClusterKatFile(t *testing.T, content string) string {
t.Helper()
tmpFile, err := ioutil.TempFile(t.TempDir(), "cluster.*.kat")
tmpFile, err := os.CreateTemp(t.TempDir(), "cluster.*.kat")
if err != nil {
t.Fatalf("Failed to create temp file: %v", err)
}
@ -144,8 +141,8 @@ spec:
}
func TestSetClusterConfigDefaults(t *testing.T) {
config := &v1alpha1.ClusterConfiguration{
Spec: &v1alpha1.ClusterConfigurationSpec{},
config := &pb.ClusterConfiguration{
Spec: &pb.ClusterConfigurationSpec{},
}
SetClusterConfigDefaults(config)
@ -184,8 +181,8 @@ func TestSetClusterConfigDefaults(t *testing.T) {
}
// Test NodeLossTimeoutSeconds derivation
configWithTick := &v1alpha1.ClusterConfiguration{
Spec: &v1alpha1.ClusterConfigurationSpec{AgentTickSeconds: 10},
configWithTick := &pb.ClusterConfiguration{
Spec: &pb.ClusterConfigurationSpec{AgentTickSeconds: 10},
}
SetClusterConfigDefaults(configWithTick)
if configWithTick.Spec.NodeLossTimeoutSeconds != 40 { // 10 * 4
@ -194,48 +191,53 @@ func TestSetClusterConfigDefaults(t *testing.T) {
}
func TestValidateClusterConfiguration_InvalidValues(t *testing.T) {
baseValidSpec := func() *v1alpha1.ClusterConfigurationSpec {
return &v1alpha1.ClusterConfigurationSpec{
ClusterCidr: "10.0.0.0/16",
ServiceCidr: "10.1.0.0/16",
NodeSubnetBits: 8,
ClusterDomain: "test.local",
AgentPort: 10250,
ApiPort: 10251,
EtcdPeerPort: 2380,
EtcdClientPort: 2379,
VolumeBasePath: "/var/lib/kat/volumes",
BackupPath: "/var/lib/kat/backups",
BackupIntervalMinutes: 30,
AgentTickSeconds: 15,
NodeLossTimeoutSeconds:60,
baseValidSpec := func() *pb.ClusterConfigurationSpec {
return &pb.ClusterConfigurationSpec{
ClusterCidr: "10.0.0.0/16",
ServiceCidr: "10.1.0.0/16",
NodeSubnetBits: 8,
ClusterDomain: "test.local",
AgentPort: 10250,
ApiPort: 10251,
EtcdPeerPort: 2380,
EtcdClientPort: 2379,
VolumeBasePath: "/var/lib/kat/volumes",
BackupPath: "/var/lib/kat/backups",
BackupIntervalMinutes: 30,
AgentTickSeconds: 15,
NodeLossTimeoutSeconds: 60,
}
}
baseValidMetadata := func() *v1alpha1.ObjectMeta {
return &v1alpha1.ObjectMeta{Name: "test"}
baseValidMetadata := func() *pb.ObjectMeta {
return &pb.ObjectMeta{Name: "test"}
}
tests := []struct {
name string
mutator func(cfg *v1alpha1.ClusterConfiguration)
mutator func(cfg *pb.ClusterConfiguration)
wantErr string
}{
{"invalid clusterCIDR", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.ClusterCidr = "invalid" }, "invalid spec.clusterCIDR"},
{"invalid serviceCIDR", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.ServiceCidr = "invalid" }, "invalid spec.serviceCIDR"},
{"invalid agentPort low", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.AgentPort = 0 }, "invalid port for agentPort"},
{"invalid agentPort high", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.AgentPort = 70000 }, "invalid port for agentPort"},
{"port conflict", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.ApiPort = cfg.Spec.AgentPort }, "port conflict"},
{"invalid nodeSubnetBits low", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.NodeSubnetBits = 0 }, "invalid spec.nodeSubnetBits"},
{"invalid nodeSubnetBits high", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.NodeSubnetBits = 32 }, "invalid spec.nodeSubnetBits"},
{"invalid nodeSubnetBits vs clusterCIDR", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.ClusterCidr = "10.0.0.0/28"; cfg.Spec.NodeSubnetBits = 8 }, "results in an invalid subnet size"},
{"invalid agentTickSeconds", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.AgentTickSeconds = 0 }, "agentTickSeconds must be positive"},
{"invalid nodeLossTimeoutSeconds", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.NodeLossTimeoutSeconds = 0 }, "nodeLossTimeoutSeconds must be positive"},
{"nodeLoss < agentTick", func(cfg *v1alpha1.ClusterConfiguration) { cfg.Spec.NodeLossTimeoutSeconds = cfg.Spec.AgentTickSeconds - 1 }, "nodeLossTimeoutSeconds must be greater"},
{"invalid clusterCIDR", func(cfg *pb.ClusterConfiguration) { cfg.Spec.ClusterCidr = "invalid" }, "invalid spec.clusterCIDR"},
{"invalid serviceCIDR", func(cfg *pb.ClusterConfiguration) { cfg.Spec.ServiceCidr = "invalid" }, "invalid spec.serviceCIDR"},
{"invalid agentPort low", func(cfg *pb.ClusterConfiguration) { cfg.Spec.AgentPort = 0 }, "invalid port for agentPort"},
{"invalid agentPort high", func(cfg *pb.ClusterConfiguration) { cfg.Spec.AgentPort = 70000 }, "invalid port for agentPort"},
{"port conflict", func(cfg *pb.ClusterConfiguration) { cfg.Spec.ApiPort = cfg.Spec.AgentPort }, "port conflict"},
{"invalid nodeSubnetBits low", func(cfg *pb.ClusterConfiguration) { cfg.Spec.NodeSubnetBits = 0 }, "invalid spec.nodeSubnetBits"},
{"invalid nodeSubnetBits high", func(cfg *pb.ClusterConfiguration) { cfg.Spec.NodeSubnetBits = 32 }, "invalid spec.nodeSubnetBits"},
{"invalid nodeSubnetBits vs clusterCIDR", func(cfg *pb.ClusterConfiguration) {
cfg.Spec.ClusterCidr = "10.0.0.0/28"
cfg.Spec.NodeSubnetBits = 8
}, "results in an invalid subnet size"},
{"invalid agentTickSeconds", func(cfg *pb.ClusterConfiguration) { cfg.Spec.AgentTickSeconds = 0 }, "agentTickSeconds must be positive"},
{"invalid nodeLossTimeoutSeconds", func(cfg *pb.ClusterConfiguration) { cfg.Spec.NodeLossTimeoutSeconds = 0 }, "nodeLossTimeoutSeconds must be positive"},
{"nodeLoss < agentTick", func(cfg *pb.ClusterConfiguration) {
cfg.Spec.NodeLossTimeoutSeconds = cfg.Spec.AgentTickSeconds - 1
}, "nodeLossTimeoutSeconds must be greater"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
config := &v1alpha1.ClusterConfiguration{Metadata: baseValidMetadata(), Spec: baseValidSpec()}
config := &pb.ClusterConfiguration{Metadata: baseValidMetadata(), Spec: baseValidSpec()}
tt.mutator(config)
err := ValidateClusterConfiguration(config)
if err == nil {

23
internal/config/types.go Normal file
View File

@ -0,0 +1,23 @@
// File: internal/config/types.go
package config
// For Phase 0, we will primarily use the generated protobuf types
// (e.g., *v1alpha1.ClusterConfiguration) directly.
// This file can hold auxiliary types or constants related to config parsing if needed later.
const (
DefaultClusterDomain = "kat.cluster.local"
DefaultAgentPort = 9116
DefaultApiPort = 9115
DefaultEtcdPeerPort = 2380
DefaultEtcdClientPort = 2379
DefaultVolumeBasePath = "/var/lib/kat/volumes"
DefaultBackupPath = "/var/lib/kat/backups"
DefaultBackupIntervalMins = 30
DefaultAgentTickSeconds = 15
DefaultNodeLossTimeoutSec = 60 // DefaultNodeLossTimeoutSeconds = DefaultAgentTickSeconds * 4 (example logic)
DefaultNodeSubnetBits = 7 // yields /23 from /16, or /31 from /24 etc. (5 bits for /29, 7 for /25)
// RFC says 7 for /23 from /16. This means 2^(32-16-7) = 2^9 = 512 IPs per node subnet.
// If nodeSubnetBits means bits for the node portion *within* the host part of clusterCIDR:
// e.g. /16 -> 16 host bits. If nodeSubnetBits = 7, then node subnet is / (16+7) = /23.
)

87
internal/utils/tar.go Normal file
View File

@ -0,0 +1,87 @@
package utils
import (
"archive/tar"
"compress/gzip"
"fmt"
"io"
"path/filepath"
"strings"
)
const maxQuadletFileSize = 1 * 1024 * 1024 // 1MB limit per file in tarball
const maxTotalQuadletSize = 5 * 1024 * 1024 // 5MB limit for total uncompressed size
const maxQuadletFiles = 20 // Max number of files in a quadlet bundle
// UntarQuadlets unpacks a tar.gz stream in memory and returns a map of fileName -> fileContent.
// It performs basic validation on file names and sizes.
func UntarQuadlets(reader io.Reader) (map[string][]byte, error) {
gzr, err := gzip.NewReader(reader)
if err != nil {
return nil, fmt.Errorf("failed to create gzip reader: %w", err)
}
defer gzr.Close()
tr := tar.NewReader(gzr)
files := make(map[string][]byte)
var totalSize int64
fileCount := 0
for {
header, err := tr.Next()
if err == io.EOF {
break // End of archive
}
if err != nil {
return nil, fmt.Errorf("failed to read tar header: %w", err)
}
// Basic security checks
if strings.Contains(header.Name, "..") {
return nil, fmt.Errorf("invalid file path in tar: %s (contains '..')", header.Name)
}
// Ensure files are *.kat and are not in subdirectories within the tarball
// The Quadlet concept implies a flat directory of *.kat files.
if filepath.Dir(header.Name) != "." && filepath.Dir(header.Name) != "" {
return nil, fmt.Errorf("invalid file path in tar: %s (subdirectories are not allowed for Quadlet files)", header.Name)
}
if !strings.HasSuffix(strings.ToLower(header.Name), ".kat") {
return nil, fmt.Errorf("invalid file type in tar: %s (only .kat files are allowed)", header.Name)
}
switch header.Typeflag {
case tar.TypeReg: // Regular file
fileCount++
if fileCount > maxQuadletFiles {
return nil, fmt.Errorf("too many files in quadlet bundle; limit %d", maxQuadletFiles)
}
if header.Size > maxQuadletFileSize {
return nil, fmt.Errorf("file %s in tar is too large: %d bytes (max %d)", header.Name, header.Size, maxQuadletFileSize)
}
totalSize += header.Size
if totalSize > maxTotalQuadletSize {
return nil, fmt.Errorf("total size of files in tar is too large (max %d MB)", maxTotalQuadletSize/(1024*1024))
}
content, err := io.ReadAll(tr)
if err != nil {
return nil, fmt.Errorf("failed to read file content for %s from tar: %w", header.Name, err)
}
if int64(len(content)) != header.Size {
return nil, fmt.Errorf("file %s in tar has inconsistent size: header %d, read %d", header.Name, header.Size, len(content))
}
files[header.Name] = content
case tar.TypeDir: // Directory
// Directories are ignored; we expect a flat structure of .kat files.
continue
default:
// Symlinks, char devices, etc. are not allowed.
return nil, fmt.Errorf("unsupported file type in tar for %s: typeflag %c", header.Name, header.Typeflag)
}
}
if len(files) == 0 {
return nil, fmt.Errorf("no .kat files found in the provided archive")
}
return files, nil
}

View File

@ -5,7 +5,6 @@ import (
"bytes"
"compress/gzip"
"io"
"os"
"path/filepath"
"strings"
"testing"
@ -132,7 +131,7 @@ func TestUntarQuadlets_FileTooLarge(t *testing.T) {
func TestUntarQuadlets_TotalSizeTooLarge(t *testing.T) {
numFiles := (maxTotalQuadletSize / maxQuadletFileSize) + 2
fileSize := maxQuadletFileSize / 2
inputFiles := make(map[string]string)
content := strings.Repeat("a", int(fileSize))
for i := 0; i < int(numFiles); i++ {
@ -150,18 +149,18 @@ func TestUntarQuadlets_TotalSizeTooLarge(t *testing.T) {
}
func TestUntarQuadlets_TooManyFiles(t *testing.T) {
inputFiles := make(map[string]string)
for i := 0; i <= maxQuadletFiles; i++ {
inputFiles[filepath.Join(".", "file"+string(rune(i+'a'))+".kat")] = "content"
}
reader := createTestTarGz(t, inputFiles, nil)
_, err := UntarQuadlets(reader)
if err == nil {
t.Fatal("UntarQuadlets() with too many files did not return an error")
}
if !strings.Contains(err.Error(), "too many files in quadlet bundle") {
t.Errorf("Expected 'too many files' error, got: %v", err)
}
inputFiles := make(map[string]string)
for i := 0; i <= maxQuadletFiles; i++ {
inputFiles[filepath.Join(".", "file"+string(rune(i+'a'))+".kat")] = "content"
}
reader := createTestTarGz(t, inputFiles, nil)
_, err := UntarQuadlets(reader)
if err == nil {
t.Fatal("UntarQuadlets() with too many files did not return an error")
}
if !strings.Contains(err.Error(), "too many files in quadlet bundle") {
t.Errorf("Expected 'too many files' error, got: %v", err)
}
}
func TestUntarQuadlets_UnsupportedFileType(t *testing.T) {

47
scripts/gen-proto.sh Executable file
View File

@ -0,0 +1,47 @@
#!/bin/bash
# File: scripts/gen-proto.sh
set -xe
# Find protoc-gen-go
PROTOC_GEN_GO_PATH=""
if command -v protoc-gen-go &> /dev/null; then
PROTOC_GEN_GO_PATH=$(command -v protoc-gen-go)
elif [ -f "$(go env GOBIN)/protoc-gen-go" ]; then
PROTOC_GEN_GO_PATH="$(go env GOBIN)/protoc-gen-go"
elif [ -f "$(go env GOPATH)/bin/protoc-gen-go" ]; then
PROTOC_GEN_GO_PATH="$(go env GOPATH)/bin/protoc-gen-go"
else
echo "protoc-gen-go not found. Please run:"
echo "go install google.golang.org/protobuf/cmd/protoc-gen-go"
echo "And ensure GOBIN or GOPATH/bin is in your PATH."
exit 1
fi
# Project root assumed to be parent of 'scripts' directory
PROJECT_ROOT="$( cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd )"
API_DIR="${PROJECT_ROOT}/api/v1alpha1"
# Output generated code directly into the api/v1alpha1 directory, alongside kat.proto
# This is a common pattern and simplifies imports.
# The go_package option in kat.proto already points here.
OUT_DIR="${API_DIR}"
# Ensure output directory exists (it should, it's the same as API_DIR)
mkdir -p "$OUT_DIR"
echo "Generating Go code from Protobuf definitions..."
protoc --proto_path="${API_DIR}" \
--plugin="protoc-gen-go=${PROTOC_GEN_GO_PATH}" \
--go_out="${OUT_DIR}" --go_opt=paths=source_relative \
"${API_DIR}/kat.proto"
echo "Protobuf Go code generated in ${OUT_DIR}"
# Optional: Generate gRPC stubs if/when you add services
# PROTOC_GEN_GO_GRPC_PATH="" # Similar logic to find protoc-gen-go-grpc
# go install google.golang.org/grpc/cmd/protoc-gen-go-grpc
# protoc --proto_path="${API_DIR}" \
# --plugin="protoc-gen-go=${PROTOC_GEN_GO_PATH}" \
# --plugin="protoc-gen-go-grpc=${PROTOC_GEN_GO_GRPC_PATH}" \
# --go_out="${OUT_DIR}" --go_opt=paths=source_relative \
# --go-grpc_out="${OUT_DIR}" --go-grpc_opt=paths=source_relative \
# "${API_DIR}/kat.proto"