En las aplicaciones modernas, la conectividad continua es clave, especialmente para las aplicaciones móviles que dependen de servicios backend. En este blog, veremos una solución basada en Python que supervisa el estado de los servidores de servicios de aplicaciones y, si es necesario, conmuta automáticamente a un servidor secundario. Este código de ejemplo utiliza comprobaciones de estado HTTP y puntos finales de conexión WebSocket para garantizar que la aplicación siempre se conecte a un servicio en buen estado.
Visión general
La solución implica dos tipos de puntos finales:
-
- URL de chequeo
- Estos puntos finales (por ejemplo
https://.../_ping) se sondean mediante HTTP CABEZA peticiones. - Determinan si el servidor de servicios de aplicaciones está en buen estado.
- Estos puntos finales (por ejemplo
- Puntos finales de conexión
- Se trata de las direcciones URL de WebSocket (p. ej,
wss://.../primary) que tu aplicación utiliza para interactuar con el backend. - El punto final de conexión activo se actualiza en función de los resultados de la comprobación de estado.
- Se trata de las direcciones URL de WebSocket (p. ej,
- URL de chequeo
Si la comprobación de estado del servidor primario falla consecutivamente, la lógica de conmutación por error cambiará la conexión de la aplicación al servidor secundario.
El código en detalle
A continuación se muestra el código completo con comentarios en línea y explicaciones detalladas:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
importar registro importar roscado importar solicita de tiempo importar dormir # Configura el registro para mostrar mensajes con fecha y hora en el nivel INFO. registro.basicConfig(nivel=registro.INFO, formato='%(asctime)s %(levelname)s: %(mensaje)s') # -------------------------------------- # Health Check URLs (App Service Servers) # -------------------------------------- # Estas URL se utilizan para comprobar la salud de los servidores enviando peticiones HEAD. health_check_urls = { "primario": "https://XXXXXXXXXXXXXX.apps.cloud.couchbase.com:4984/_ping", "secundario": "https://XXXXXXXXXXXXXX.apps.cloud.couchbase.com:4984/_ping" } # ------------------------------------- Puntos finales de conexión # (URL de WebSocket) # ------------------------------------- # Estos puntos finales son los que su aplicación utiliza realmente para las conexiones. urls_de_conexión = { "primario": "wss://XXXXXXXXXXXX.apps.cloud.couchbase.com:4984/primary", "secundario": "wss://XXXXXXXXXXXX.apps.cloud.ucouchbase.com:4984/secondary" } # La variable `active_cluster` rastrea qué servidor está activo actualmente. cluster_activo = "primario" # Esta variable contiene la URL WebSocket real utilizada por su aplicación. active_connection_url = urls_de_conexión[cluster_activo] def is_cluster_healthy(url): """ Realiza un chequeo utilizando una petición HEAD contra la URL proporcionada. Devuelve True si el estado de la respuesta es 200; en caso contrario, devuelve False. Registra el código de estado y las cabeceras para la resolución de problemas. """ pruebe: respuesta = solicita.cabeza(url, tiempo de espera=5) registro.información(f"Respuesta de chequeo para {url}") registro.información(f" Código de estado: {response.status_code}") registro.información(" Cabeceras:") para cabecera, valor en respuesta.cabeceras.artículos(): registro.información(f" {cabecera}: {valor}") si respuesta.código_estado == 200: registro.información(f"¡{url} es saludable!") devolver Verdadero si no: registro.advertencia(f"{url} podría ser insalubre o inalcanzable".) devolver Falso excepto solicita.excepciones.RequestException como e: registro.error(f"Fallo en la comprobación sanitaria de {url}: {e}") devolver Falso def chequeo_trabajador(): """ Un trabajador en segundo plano que comprueba la salud del servidor activo cada 3 segundos. Si el servidor activo falla las comprobaciones de estado durante más de 9 veces consecutivas, el trabajador intenta cambiar al otro servidor. """ global activo_grupo global conexión_activa_url fracasos_consecutivos = 0 mientras que Verdadero: dormir(3) # Espere 3 segundos entre comprobaciones. # Utilice el punto final de comprobación de estado HTTP para el clúster activo. current_health_url = health_check_urls[cluster_activo] registro.información(f"Comprobación de salud: Comprobando {cluster_activo} en {current_health_url}...") si is_cluster_healthy(current_health_url): fracasos_consecutivos = 0 # Reinicia el contador si está sano. si no: fracasos_consecutivos += 1 registro.advertencia(f"{active_cluster} health check failed {consecutive_failures} time(s)".) # Si los fallos superan los 9 intentos consecutivos, intente conmutar por error. si fracasos_consecutivos > 9: registro.error(f"{cluster_activo} se considera caído. Intentando conmutar por error...") # Determinar el nuevo cluster activo. nuevo_cluster = "secundario" si cluster_activo == "primario" si no "primario" nueva_health_url = health_check_urls[nuevo_cluster] # Compruebe si el nuevo clúster está en buen estado. si is_cluster_healthy(nueva_health_url): cluster_activo = nuevo_grupo # Actualizar el punto final de la conexión. active_connection_url = urls_de_conexión[nuevo_cluster] registro.advertencia(f"Cambiado cluster activo a {nuevo_cluster}".) registro.advertencia(f"Nuevo punto final de conexión WebSocket: {active_connection_url}") si no: registro.crítico("¡Ambos grupos parecen haber caído!") fracasos_consecutivos = 0 # Reinicia el contador de fallos después del intento. def principal(): """ Función principal para iniciar el health-check worker en un subproceso en segundo plano. Mantiene el script en ejecución indefinidamente hasta que se interrumpe. """ hilo = roscado.Hilo(objetivo=chequeo_trabajador, demonio=Verdadero) hilo.iniciar() registro.información("Comprobación de salud del trabajador iniciada. Pulse Ctrl+C para salir".) registro.información(f"La aplicación se conectará inicialmente a: {active_connection_url}") pruebe: mientras que Verdadero: dormir(1) # El hilo principal sigue vivo. excepto Interrupción del teclado: registro.información("Cerrando script de chequeo de salud".) si __name__ == "__main__": principal() |
Puntos técnicos clave
-
- Comprobaciones de estado de los servidores App Service:
El código separa el puntos finales del chequeo (utilizado para la supervisión) del puntos finales de conexión (utilizado por su aplicación). Esto le permite comprobar el estado del servidor de forma independiente mientras mantiene un punto final de conexión estable. - Solicitudes HTTP HEAD:
Utilizando CABEZA solicitudes a la/_pingminimiza la transferencia de datos al tiempo que proporciona códigos de estado y cabeceras para el diagnóstico. - Hilo de fondo:
Enchequeo_trabajadorse ejecuta en su propio subproceso, lo que permite una supervisión continua sin bloquear el subproceso principal de la aplicación. - Lógica de conmutación por error:
- Un contador (
fracasos_consecutivos) realiza un seguimiento de los fallos consecutivos. - Si el recuento supera un umbral establecido (9 fallos), el script intenta una conmutación por error comprobando el estado del servidor alternativo.
- Tras una comprobación de estado satisfactoria en el servidor secundario, se actualiza el punto final de conexión activo.
- Un contador (
- Registro:
El registro detallado proporciona información sobre el proceso de comprobación de estado, incluidos el estado de respuesta HTTP, las cabeceras y los eventos de conmutación por error. Esto facilita la resolución de problemas y la supervisión.
- Comprobaciones de estado de los servidores App Service:
Adaptación a su aplicación
-
- Puedes traducir y adaptar fácilmente este código a tu lenguaje de programación preferido, como Swift y Kotlin, para adaptarlo a las necesidades de tu aplicación.
- Puede integrar esta secuencia de comandos o lógica en su código móvil (iOS/Android) o un servicio backend que actualiza el activo punto final.
- Si utiliza iOS o Android, tenga en cuenta con qué frecuencia y dónde ejecuta este código. Por ejemplo, las tareas en segundo plano o las notificaciones push pueden activar comprobaciones de estado en un contexto móvil.
- Si usted tiene una arquitectura de microservicios, puede ejecutar esta lógica de conmutación por error en un pequeño servicio que expone un archivo URL activa actual a las aplicaciones móviles, para que siempre se conecten al punto final de WSS correcto.
Conclusión
Este código de ejemplo proporciona un mecanismo sencillo pero potente para garantizar la alta disponibilidad de las aplicaciones mediante la conmutación automática a un servidor de copia de seguridad cuando el servidor primario deja de estar accesible. Al separar las comprobaciones de estado de los puntos finales de conexión, la aplicación garantiza que siempre se conecta a un servidor en buen estado a través de WebSocket.
En un entorno de producción, es posible que tenga que adaptar y ampliar la lógica para adaptarla a sus requisitos específicos, las condiciones de la red y las políticas de seguridad.
Implementar esta lógica en su aplicación móvil o servicio backend puede mejorar enormemente el tiempo de actividad y la resistencia, garantizando que sus usuarios sigan conectados incluso durante interrupciones inesperadas del servicio.
