Tengo un cliente Go gRPC conectado a un servidor gRPC que se ejecuta en un módulo diferente en mi clúster k8s.
Está funcionando bien, recibiendo y procesando solicitudes.
Ahora me pregunto cuál es la mejor manera de implementar la resiliencia en caso de que el módulo del servidor gRPC se recicle.
Por lo que puedo asegurar, el código de clientconn.go debería manejar la reconexión automáticamente, pero no puedo hacer que funcione y me temo que mi implementación es incorrecta en primera instancia.
Código de llamada desde principal:
go func() { if err := gRPCClient.ProcessRequests(); err != nil { log.Error("Error while processing Requests") //do something here?? } }()Mi código en el módulo contenedor gRPCClient:
func (grpcclient *gRPCClient) ProcessRequests() error { defer grpcclient.Close() for { request, err := reqclient.stream.Recv() log.Info("Request received") if err == io.EOF { break } if err != nil { //when pod is recycled, this is what's hit with err: //rpc error: code = Unavailable desc = transport is closing" //what is the correct pattern for recovery here so that we can await connection //and continue processing requests once more? //should I return err here and somehow restart the ProcessRequests() go routine in the //main funcition? break } else { //the happy path //code block to process any requests that are received } } return nil } func (reqclient *RequestClient) Close() { //this is called soon after the conneciton drops reqclient.conn.Close() }EDITAR: Emin Laletovic respondió a mi pregunta elegantemente a continuación y lo obtiene la mayor parte del camino. Tuve que hacer algunos cambios en la función waitUntilReady:
func (grpcclient *gRPCClient) waitUntilReady() bool { ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second) //define how long you want to wait for connection to be restored before giving up defer cancel() currentState := grpcclient.conn.GetState() stillConnecting := true for currentState != connectivity.Ready && stillConnecting { //will return true when state has changed from thisState, false if timeout stillConnecting = grpcclient.conn.WaitForStateChange(ctx, currentState) currentState = grpcclient.conn.GetState() log.WithFields(log.Fields{"state: ": currentState, "timeout": timeoutDuration}).Info("Attempting reconnection. State has changed to:") } if stillConnecting == false { log.Error("Connection attempt has timed out.") return false } return true }clientconn.go maneja automáticamente la conexión RPC, pero eso no significa que la transmisión también se maneja automáticamente.
La transmisión, una vez interrumpida, ya sea por la interrupción de la conexión RPC o por alguna otra razón, no puede volver a conectarse automáticamente y debe obtener una nueva transmisión del servidor una vez que la conexión RPC se haya restablecido.
El pseudocódigo para esperar que la conexión RPC esté en el estado READY y establecer una nueva transmisión podría verse así:
func (grpcclient *gRPCClient) ProcessRequests() error { defer grpcclient.Close() go grpcclient.process() for { select { case <- grpcclient.reconnect: if !grpcclient.waitUntilReady() { return errors.New("failed to establish a connection within the defined timeout") } go grpcclient.process() case <- grpcclient.done: return nil } } } func (grpcclient *gRPCClient) process() { reqclient := GetStream() //always get a new stream for { request, err := reqclient.stream.Recv() log.Info("Request received") if err == io.EOF { grpcclient.done <- true return } if err != nil { grpcclient.reconnect <- true return } else { //the happy path //code block to process any requests that are received } } } func (grpcclient *gRPCClient) waitUntilReady() bool { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) //define how long you want to wait for connection to be restored before giving up defer cancel() return grpcclient.conn.WaitForStateChange(ctx, conectivity.Ready) }