refactor: Align Epic 0 & Epic 1 with true microservices architecture
Refactor core kernel and infrastructure to support true microservices architecture where services are independently deployable. Phase 1: Core Kernel Cleanup - Remove database provider from CoreModule (services create their own) - Update ProvideHealthRegistry to not depend on database - Add schema support to database client (NewClientWithSchema) - Update main entry point to remove database dependency - Core kernel now provides only: config, logger, error bus, health, metrics, tracer, service registry Phase 2: Service Registry Implementation - Create ServiceRegistry interface (pkg/registry/registry.go) - Implement Consul registry (internal/registry/consul/consul.go) - Add Consul dependency (github.com/hashicorp/consul/api) - Add registry configuration to config/default.yaml - Add ProvideServiceRegistry() to DI container Phase 3: Service Client Interfaces - Create service client interfaces: - pkg/services/auth.go - AuthServiceClient - pkg/services/identity.go - IdentityServiceClient - pkg/services/authz.go - AuthzServiceClient - pkg/services/audit.go - AuditServiceClient - Create ServiceClientFactory (internal/client/factory.go) - Create stub gRPC client implementations (internal/client/grpc/) - Add ProvideServiceClientFactory() to DI container Phase 4: gRPC Service Definitions - Create proto files for all core services: - api/proto/auth.proto - api/proto/identity.proto - api/proto/authz.proto - api/proto/audit.proto - Add generate-proto target to Makefile Phase 5: API Gateway Implementation - Create API Gateway service entry point (cmd/api-gateway/main.go) - Create Gateway implementation (services/gateway/gateway.go) - Add gateway configuration to config/default.yaml - Gateway registers with Consul and routes requests to backend services All code compiles successfully. Core services (Auth, Identity, Authz, Audit) will be implemented in Epic 2 using these foundations.
This commit is contained in:
198
internal/registry/consul/consul.go
Normal file
198
internal/registry/consul/consul.go
Normal file
@@ -0,0 +1,198 @@
|
||||
// Package consul provides Consul-based service registry implementation.
|
||||
package consul
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.dcentral.systems/toolz/goplt/pkg/registry"
|
||||
consulapi "github.com/hashicorp/consul/api"
|
||||
)
|
||||
|
||||
// ConsulRegistry implements ServiceRegistry using Consul.
|
||||
type ConsulRegistry struct {
|
||||
client *consulapi.Client
|
||||
config *Config
|
||||
}
|
||||
|
||||
// Config holds Consul registry configuration.
|
||||
type Config struct {
|
||||
Address string // Consul agent address (e.g., "localhost:8500")
|
||||
Datacenter string // Consul datacenter
|
||||
Scheme string // "http" or "https"
|
||||
HealthCheck HealthCheckConfig
|
||||
}
|
||||
|
||||
// HealthCheckConfig holds health check configuration.
|
||||
type HealthCheckConfig struct {
|
||||
Interval time.Duration // Health check interval
|
||||
Timeout time.Duration // Health check timeout
|
||||
DeregisterAfter time.Duration // Time to wait before deregistering unhealthy service
|
||||
HTTP string // HTTP health check endpoint (e.g., "/healthz")
|
||||
}
|
||||
|
||||
// NewRegistry creates a new Consul-based service registry.
|
||||
func NewRegistry(cfg Config) (*ConsulRegistry, error) {
|
||||
consulConfig := consulapi.DefaultConfig()
|
||||
if cfg.Address != "" {
|
||||
consulConfig.Address = cfg.Address
|
||||
}
|
||||
if cfg.Datacenter != "" {
|
||||
consulConfig.Datacenter = cfg.Datacenter
|
||||
}
|
||||
if cfg.Scheme != "" {
|
||||
consulConfig.Scheme = cfg.Scheme
|
||||
}
|
||||
|
||||
client, err := consulapi.NewClient(consulConfig)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create Consul client: %w", err)
|
||||
}
|
||||
|
||||
return &ConsulRegistry{
|
||||
client: client,
|
||||
config: &cfg,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Register registers a service instance with Consul.
|
||||
func (r *ConsulRegistry) Register(ctx context.Context, service *registry.ServiceInstance) error {
|
||||
registration := &consulapi.AgentServiceRegistration{
|
||||
ID: service.ID,
|
||||
Name: service.Name,
|
||||
Address: service.Address,
|
||||
Port: service.Port,
|
||||
Tags: service.Tags,
|
||||
Meta: service.Metadata,
|
||||
}
|
||||
|
||||
// Add health check if configured
|
||||
if r.config.HealthCheck.HTTP != "" {
|
||||
healthCheckURL := fmt.Sprintf("http://%s:%d%s", service.Address, service.Port, r.config.HealthCheck.HTTP)
|
||||
registration.Check = &consulapi.AgentServiceCheck{
|
||||
HTTP: healthCheckURL,
|
||||
Interval: r.config.HealthCheck.Interval.String(),
|
||||
Timeout: r.config.HealthCheck.Timeout.String(),
|
||||
DeregisterCriticalServiceAfter: r.config.HealthCheck.DeregisterAfter.String(),
|
||||
}
|
||||
}
|
||||
|
||||
return r.client.Agent().ServiceRegister(registration)
|
||||
}
|
||||
|
||||
// Deregister removes a service instance from Consul.
|
||||
func (r *ConsulRegistry) Deregister(ctx context.Context, serviceID string) error {
|
||||
return r.client.Agent().ServiceDeregister(serviceID)
|
||||
}
|
||||
|
||||
// Discover returns all healthy instances of a service.
|
||||
func (r *ConsulRegistry) Discover(ctx context.Context, serviceName string) ([]*registry.ServiceInstance, error) {
|
||||
services, _, err := r.client.Health().Service(serviceName, "", true, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to discover service %s: %w", serviceName, err)
|
||||
}
|
||||
|
||||
instances := make([]*registry.ServiceInstance, 0, len(services))
|
||||
for _, service := range services {
|
||||
instances = append(instances, ®istry.ServiceInstance{
|
||||
ID: service.Service.ID,
|
||||
Name: service.Service.Service,
|
||||
Address: service.Service.Address,
|
||||
Port: service.Service.Port,
|
||||
Tags: service.Service.Tags,
|
||||
Metadata: service.Service.Meta,
|
||||
})
|
||||
}
|
||||
|
||||
return instances, nil
|
||||
}
|
||||
|
||||
// Watch returns a channel that receives updates when service instances change.
|
||||
func (r *ConsulRegistry) Watch(ctx context.Context, serviceName string) (<-chan []*registry.ServiceInstance, error) {
|
||||
updates := make(chan []*registry.ServiceInstance, 10)
|
||||
|
||||
go func() {
|
||||
defer close(updates)
|
||||
|
||||
lastIndex := uint64(0)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
services, meta, err := r.client.Health().Service(serviceName, "", true, &consulapi.QueryOptions{
|
||||
WaitIndex: lastIndex,
|
||||
WaitTime: 10 * time.Second,
|
||||
})
|
||||
if err != nil {
|
||||
// Log error and continue
|
||||
continue
|
||||
}
|
||||
|
||||
if meta.LastIndex != lastIndex {
|
||||
instances := make([]*registry.ServiceInstance, 0, len(services))
|
||||
for _, service := range services {
|
||||
instances = append(instances, ®istry.ServiceInstance{
|
||||
ID: service.Service.ID,
|
||||
Name: service.Service.Service,
|
||||
Address: service.Service.Address,
|
||||
Port: service.Service.Port,
|
||||
Tags: service.Service.Tags,
|
||||
Metadata: service.Service.Meta,
|
||||
})
|
||||
}
|
||||
|
||||
select {
|
||||
case updates <- instances:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
lastIndex = meta.LastIndex
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return updates, nil
|
||||
}
|
||||
|
||||
// Health returns the health status of a service instance.
|
||||
func (r *ConsulRegistry) Health(ctx context.Context, serviceID string) (*registry.HealthStatus, error) {
|
||||
entries, _, err := r.client.Health().Service(serviceID, "", false, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get health for service %s: %w", serviceID, err)
|
||||
}
|
||||
|
||||
if len(entries) == 0 {
|
||||
return ®istry.HealthStatus{
|
||||
ServiceID: serviceID,
|
||||
Status: "unknown",
|
||||
Message: "service not found",
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Check health status from service entry checks
|
||||
status := "healthy"
|
||||
message := "all checks passing"
|
||||
|
||||
// Get the first entry (should be the service instance)
|
||||
entry := entries[0]
|
||||
for _, check := range entry.Checks {
|
||||
if check.Status == consulapi.HealthCritical {
|
||||
status = "critical"
|
||||
message = check.Output
|
||||
break
|
||||
} else if check.Status == consulapi.HealthWarning {
|
||||
status = "unhealthy"
|
||||
message = check.Output
|
||||
}
|
||||
}
|
||||
|
||||
return ®istry.HealthStatus{
|
||||
ServiceID: serviceID,
|
||||
Status: status,
|
||||
Message: message,
|
||||
}, nil
|
||||
}
|
||||
Reference in New Issue
Block a user