mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 08:09:27 +00:00
+2
8cd0683a0b
8cd0683a0b
# WebSocket v2 Architecture
## NOTE: Open Issues:
- [ ] `session_drain` from server is experimental and not ready for
prod. Instead, the client maintains ongoing `trace_id`'s and when cookie
expires it will retry search for all the ongoing searches.
## Overview
Implements websocket v2 with the main goal to establish `Client-Router`
connections `per tab` and `Router-Querier` `persistent` connections.
## New Env (Default):
```sh
ZO_WEBSOCKET_CHECK_COOKIE_EXPIRY=true
ZO_WEBSOCKET_HEALTH_CHECK_INTERVAL_SECS=60
ZO_WEBSOCKET_IS_SESSION_DRAIN_ENABLED=false # This is experimental, not meant for prod
```
## High-Level Flow
1. Client establishes WebSocket connection with the router.
2. Router creates a session.
3. Router maintains persistent connections to all queriers.
4. Client requests are routed to the appropriate querier using
consistent hashing.
5. Responses are routed back using `trace_id` mappings.
6. Health checks maintain connection validity.
7. Failed connections trigger automatic cleanup and reconnection.
8. **Internal Communication Validation Using:**
- **8a.** Internal gRPC tokens
- **8b.** IP validation
- **8c.** WebSocket upgrade headers
## Connection Lifecycle
```mermaid
sequenceDiagram
participant Client
participant Router
participant ConnectionPool
participant QuerierConn
participant Querier
participant Cluster
Client->>Router: WS Connection Request
Router->>SessionManager: register_client(client_id)
Note over Router,SessionManager: Creates channels:<br/>response_tx/rx<br/>disconnect_tx/rx
par Handle Incoming Messages
Router->>Router: spawn handle_incoming task
and Handle Outgoing Messages
Router->>Router: spawn handle_outgoing task
end
Client->>Router: WS Client Event
Router->>SessionManager: get_querier_for_trace(trace_id)
alt No Existing Mapping
Router->>Cluster: get_node_from_consistent_hash(trace_id)
Router->>ConnectionPool: get_or_create_connection(querier)
alt No Existing Connection
ConnectionPool->>QuerierConn: create_connection
QuerierConn->>Querier: WS Connection Request
par Connection Tasks
QuerierConn->>QuerierConn: spawn listen_to_querier_response
QuerierConn->>QuerierConn: spawn health_check
end
Querier-->>QuerierConn: Connection Established
end
Router->>SessionManager: set_querier_for_trace(trace_id, querier)
end
Router->>QuerierConn: register_request(trace_id)
Router->>QuerierConn: send_message(event)
Querier-->>QuerierConn: Response
QuerierConn->>Router: Route Response via response_tx
Router->>Client: Forward Response
Note over Router,Client: On Disconnect/Error:<br/>1. Send disconnect signal<br/>2. Clean up sessions<br/>3. Drain if needed
```
## Connections
```mermaid
graph TB
subgraph "Client Layer"
UI1[UI Client 1]
UI2[UI Client 2]
UI3[UI Client 3]
end
subgraph "Router Layer"
R1[Router 1]
R2[Router 2]
subgraph "Router 1 Connection Pool"
CP1[Connection Pool]
QC11[Querier1 Conn]
QC12[Querier2 Conn]
QC13[Querier3 Conn]
CP1 --- QC11
CP1 --- QC12
CP1 --- QC13
end
subgraph "Router 2 Connection Pool"
CP2[Connection Pool]
QC21[Querier1 Conn]
QC22[Querier2 Conn]
QC23[Querier3 Conn]
CP2 --- QC21
CP2 --- QC22
CP2 --- QC23
end
R1 --- CP1
R2 --- CP2
end
subgraph "Querier Layer"
Q1[Querier 1]
Q2[Querier 2]
Q3[Querier 3]
end
UI1 -.->|WS| R1
UI2 -.->|WS| R1
UI3 -.->|WS| R2
QC11 ===|Persistent WS| Q1
QC12 ===|Persistent WS| Q2
QC13 ===|Persistent WS| Q3
QC21 ===|Persistent WS| Q1
QC22 ===|Persistent WS| Q2
QC23 ===|Persistent WS| Q3
classDef client fill:#f9f,stroke:#333,stroke-width:2px
classDef router fill:#bbf,stroke:#333,stroke-width:2px
classDef querier fill:#bfb,stroke:#333,stroke-width:2px
classDef pool fill:#ffd,stroke:#333,stroke-width:2px
class UI1,UI2,UI3 client
class R1,R2 router
class Q1,Q2,Q3 querier
class CP1,CP2 pool
```
## Components
```mermaid
classDiagram
class WsHandler {
+session_manager: Arc<SessionManager>
+connection_pool: Arc<QuerierConnectionPool>
+handle_connection()
+remove_querier_connection()
+_shutdown()
}
class SessionManager {
+sessions: RwAHashMap<ClientId, SessionInfo>
+mapped_queriers: RwAHashMap<QuerierName, HashSet<TraceId>>
+register_client()
+update_session_activity()
+remove_trace_id()
+unregister_client()
+set_querier_for_trace()
+get_querier_for_trace()
}
class QuerierConnectionPool {
+connections: RwAHashMap<QuerierName, Arc<QuerierConnection>>
+get_or_create_connection()
+remove_querier_connection()
+maintain_connections()
+get_active_connection()
}
class QuerierConnection {
+querier_name: QuerierName
+write: Arc<Mutex<Option<SplitSink>>>
+response_router: Arc<ResponseRouter>
+is_connected: Arc<AtomicBool>
+connect()
+disconnect()
+send_message()
+register_request()
+unregister_request()
}
class ResponseRouter {
+routes: RwAHashMap<TraceId, Sender<WsServerEvents>>
+route_response()
+flush()
+spawn_cleanup_task()
}
WsHandler --> SessionManager
WsHandler --> QuerierConnectionPool
QuerierConnectionPool --> QuerierConnection
QuerierConnection --> ResponseRouter
```
Related Issues: #6396
---------
Co-authored-by: Taiming Liu <liutaiming3@gmail.com>
Co-authored-by: Hengfei Yang <hengfei.yang@gmail.com>
Co-authored-by: Abhay Padamani <141122205+ktx-abhay@users.noreply.github.com>
Co-authored-by: Ashish Kolhe <35160958+oasisk@users.noreply.github.com>
Co-authored-by: udev <uddhav@openobserve.ai>
Co-authored-by: omkark06 <omkar@zinclabs.io>
Co-authored-by: ktx-abhay <abhay.padamani@kiara.tech>
Co-authored-by: Omkar Kesarkhane <127078473+omkarK06@users.noreply.github.com>
Co-authored-by: udev <uddhavdave1@gmail.com>
Co-authored-by: Subhra264 <chakrabortysubhradeep556@gmail.com>
Co-authored-by: ktx-vaidehi <134508096+ktx-vaidehi@users.noreply.github.com>
pytest-openobserve
This is a repo to test openobserve API endpoints.
Introduction:
The tests are running using pytest against alpha1 for now.
Idea is to parametrize it for the other endpoints like production as well.
Getting Started
- We use
ryeto manage the python version and environment - https://github.com/mitsuhiko/rye - Once installed, simply run the following steps:
rye syncrye run pytest
TODOS:
- Parametrize tests to run against different environments
- Run the tests as cron-job every few hours or daily etc. for different environments
- Send slack-notification ( use github-actions ) in case something fails on
alpha1orproductionendpoints.