// There is a Backend interface with the Execute method // Need to write a load balancer // The load balancer satisfies the Backend interface // The task of the load balancer is to accept a request and delegate it to one of the available backends // The load balancer contains a collection of backends // When executing a request, the load balancer chooses one of the backends based on some algorithm // // Algorithm Least Connection, health check and automatic recovery, thread safety via atomic, context passing for shutdown, error threshold at which we will disable the backend // Request represents an input request for Backend. type Request struct { // Payload – arbitrary request data. Payload []byte } // Response represents a response from Backend. type Response struct { // Payload – arbitrary response data. Payload []byte } // Backend – interface defining the method for executing a request type Backend interface { Execute(ctx context.Context, req Request) (Response, error) } type Balancer struct { backends []Backend } func NewBalancer(backends []Backend) *Balancer { return &Balancer{backends: backends} } func (b *Balancer) Execute(ctx context.Context, req Request) (Response, error) { // todo: implement balancer logic return Response{}, nil } type TestBackend struct { }
sobes.tech AI
Answer from AI
go import ( "context" "errors" "sync/atomic" "time" )
type backendWrapper struct { backend Backend activeConns int64 // atomic errorCount int64 // atomic healthy int32 // atomic (1 - healthy, 0 - unhealthy) lastCheckTime int64 // unix timestamp }
func (bw *backendWrapper) isHealthy() bool { return atomic.LoadInt32(&bw.healthy) == 1 }
func (bw *backendWrapper) markUnhealthy() { atomic.StoreInt32(&bw.healthy, 0) atomic.StoreInt64(&bw.errorCount, 0) }
func (bw *backendWrapper) markHealthy() { atomic.StoreInt32(&bw.healthy, 1) atomic.StoreInt64(&bw.errorCount, 0) }
func (bw *backendWrapper) incrementError() { atomic.AddInt64(&bw.errorCount, 1) }
func (bw *backendWrapper) resetError() { atomic.StoreInt64(&bw.errorCount, 0) }
func (bw *backendWrapper) incrementConns() { atomic.AddInt64(&bw.activeConns, 1) }
func (bw *backendWrapper) decrementConns() { atomic.AddInt64(&bw.activeConns, -1) }
const ( errorThreshold = 5 healthCheckInterval = 10 * time.Second )
type Balancer struct { backends []*backendWrapper }
func NewBalancer(backends []Backend) *Balancer { wrappers := make([]*backendWrapper, len(backends)) for i, b := range backends { wrappers[i] = &backendWrapper{ backend: b, healthy: 1, } } return &Balancer{backends: wrappers} }
func (b *Balancer) Execute(ctx context.Context, req Request) (Response, error) { // Run health check in background b.healthCheck()
// Select backend with the fewest active connections that is healthy
var selected *backendWrapper
minConns := int64(1<<63 - 1) // max int64
for _, bw := range b.backends {
if !bw.isHealthy() {
continue
}
conns := atomic.LoadInt64(&bw.activeConns)
if conns < minConns {
minConns = conns
selected = bw
}
}
if selected == nil {
return Response{}, errors.New("no healthy backends available")
}
selected.incrementConns()
defer selected.decrementConns()
resp, err := selected.backend.Execute(ctx, req)
if err != nil {
selected.incrementError()
if atomic.LoadInt64(&selected.errorCount) >= errorThreshold {
selected.markUnhealthy()
}
return resp, err
}
selected.resetError()
return resp, nil
}
func (b *Balancer) healthCheck() { now := time.Now().Unix() for _, bw := range b.backends { lastCheck := atomic.LoadInt64(&bw.lastCheckTime) if now-lastCheck < int64(healthCheckInterval.Seconds()) { continue } atomic.StoreInt64(&bw.lastCheckTime, now)
go func(bw *backendWrapper) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
// Try to perform a health check request
_, err := bw.backend.Execute(ctx, Request{Payload: []byte("healthcheck")})
if err != nil {
bw.markUnhealthy()
} else {
bw.markHealthy()
}
}(bw)
}
}