mirror of https://github.com/jetkvm/kvm.git
add extra logging and tune timeout settings for cloud (#167)
* chore(config): merge userConfig with defaultConfig and add a lock * chore(cloud): add extra logging and tune timeout settings
This commit is contained in:
parent
69461140e3
commit
591d512b11
63
cloud.go
63
cloud.go
|
@ -24,6 +24,18 @@ type CloudRegisterRequest struct {
|
||||||
ClientId string `json:"clientId"`
|
ClientId string `json:"clientId"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const (
|
||||||
|
// CloudWebSocketConnectTimeout is the timeout for the websocket connection to the cloud
|
||||||
|
CloudWebSocketConnectTimeout = 1 * time.Minute
|
||||||
|
// CloudAPIRequestTimeout is the timeout for cloud API requests
|
||||||
|
CloudAPIRequestTimeout = 10 * time.Second
|
||||||
|
// CloudOidcRequestTimeout is the timeout for OIDC token verification requests
|
||||||
|
// should be lower than the websocket response timeout set in cloud-api
|
||||||
|
CloudOidcRequestTimeout = 10 * time.Second
|
||||||
|
// CloudWebSocketPingInterval is the interval at which the websocket client sends ping messages to the cloud
|
||||||
|
CloudWebSocketPingInterval = 15 * time.Second
|
||||||
|
)
|
||||||
|
|
||||||
func handleCloudRegister(c *gin.Context) {
|
func handleCloudRegister(c *gin.Context) {
|
||||||
var req CloudRegisterRequest
|
var req CloudRegisterRequest
|
||||||
|
|
||||||
|
@ -44,22 +56,31 @@ func handleCloudRegister(c *gin.Context) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
resp, err := http.Post(req.CloudAPI+"/devices/token", "application/json", bytes.NewBuffer(jsonPayload))
|
client := &http.Client{Timeout: CloudAPIRequestTimeout}
|
||||||
|
|
||||||
|
apiReq, err := http.NewRequest(http.MethodPost, config.CloudURL+"/devices/token", bytes.NewBuffer(jsonPayload))
|
||||||
|
if err != nil {
|
||||||
|
c.JSON(500, gin.H{"error": "Failed to create register request: " + err.Error()})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
apiReq.Header.Set("Content-Type", "application/json")
|
||||||
|
|
||||||
|
apiResp, err := client.Do(apiReq)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
c.JSON(500, gin.H{"error": "Failed to exchange token: " + err.Error()})
|
c.JSON(500, gin.H{"error": "Failed to exchange token: " + err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
defer apiResp.Body.Close()
|
||||||
|
|
||||||
if resp.StatusCode != http.StatusOK {
|
if apiResp.StatusCode != http.StatusOK {
|
||||||
c.JSON(resp.StatusCode, gin.H{"error": "Failed to exchange token: " + resp.Status})
|
c.JSON(apiResp.StatusCode, gin.H{"error": "Failed to exchange token: " + apiResp.Status})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
var tokenResp struct {
|
var tokenResp struct {
|
||||||
SecretToken string `json:"secretToken"`
|
SecretToken string `json:"secretToken"`
|
||||||
}
|
}
|
||||||
if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil {
|
if err := json.NewDecoder(apiResp.Body).Decode(&tokenResp); err != nil {
|
||||||
c.JSON(500, gin.H{"error": "Failed to parse token response: " + err.Error()})
|
c.JSON(500, gin.H{"error": "Failed to parse token response: " + err.Error()})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
@ -70,7 +91,7 @@ func handleCloudRegister(c *gin.Context) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if config.CloudToken == "" {
|
if config.CloudToken == "" {
|
||||||
logger.Info("Starting websocket client due to adoption")
|
cloudLogger.Info("Starting websocket client due to adoption")
|
||||||
go RunWebsocketClient()
|
go RunWebsocketClient()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -122,7 +143,7 @@ func runWebsocketClient() error {
|
||||||
header := http.Header{}
|
header := http.Header{}
|
||||||
header.Set("X-Device-ID", GetDeviceID())
|
header.Set("X-Device-ID", GetDeviceID())
|
||||||
header.Set("Authorization", "Bearer "+config.CloudToken)
|
header.Set("Authorization", "Bearer "+config.CloudToken)
|
||||||
dialCtx, cancelDial := context.WithTimeout(context.Background(), time.Minute)
|
dialCtx, cancelDial := context.WithTimeout(context.Background(), CloudWebSocketConnectTimeout)
|
||||||
defer cancelDial()
|
defer cancelDial()
|
||||||
c, _, err := websocket.Dial(dialCtx, wsURL.String(), &websocket.DialOptions{
|
c, _, err := websocket.Dial(dialCtx, wsURL.String(), &websocket.DialOptions{
|
||||||
HTTPHeader: header,
|
HTTPHeader: header,
|
||||||
|
@ -131,15 +152,15 @@ func runWebsocketClient() error {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer c.CloseNow()
|
defer c.CloseNow()
|
||||||
logger.Infof("WS connected to %v", wsURL.String())
|
cloudLogger.Infof("websocket connected to %s", wsURL.String())
|
||||||
runCtx, cancelRun := context.WithCancel(context.Background())
|
runCtx, cancelRun := context.WithCancel(context.Background())
|
||||||
defer cancelRun()
|
defer cancelRun()
|
||||||
go func() {
|
go func() {
|
||||||
for {
|
for {
|
||||||
time.Sleep(15 * time.Second)
|
time.Sleep(CloudWebSocketPingInterval)
|
||||||
err := c.Ping(runCtx)
|
err := c.Ping(runCtx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Warnf("websocket ping error: %v", err)
|
cloudLogger.Warnf("websocket ping error: %v", err)
|
||||||
cancelRun()
|
cancelRun()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
@ -157,24 +178,30 @@ func runWebsocketClient() error {
|
||||||
var req WebRTCSessionRequest
|
var req WebRTCSessionRequest
|
||||||
err = json.Unmarshal(msg, &req)
|
err = json.Unmarshal(msg, &req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Warnf("unable to parse ws message: %v", string(msg))
|
cloudLogger.Warnf("unable to parse ws message: %v", string(msg))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
cloudLogger.Infof("new session request: %v", req.OidcGoogle)
|
||||||
|
cloudLogger.Tracef("session request info: %v", req)
|
||||||
|
|
||||||
err = handleSessionRequest(runCtx, c, req)
|
err = handleSessionRequest(runCtx, c, req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.Infof("error starting new session: %v", err)
|
cloudLogger.Infof("error starting new session: %v", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSessionRequest) error {
|
func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSessionRequest) error {
|
||||||
oidcCtx, cancelOIDC := context.WithTimeout(ctx, time.Minute)
|
oidcCtx, cancelOIDC := context.WithTimeout(ctx, CloudOidcRequestTimeout)
|
||||||
defer cancelOIDC()
|
defer cancelOIDC()
|
||||||
provider, err := oidc.NewProvider(oidcCtx, "https://accounts.google.com")
|
provider, err := oidc.NewProvider(oidcCtx, "https://accounts.google.com")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println("Failed to initialize OIDC provider:", err)
|
_ = wsjson.Write(context.Background(), c, gin.H{
|
||||||
|
"error": fmt.Sprintf("failed to initialize OIDC provider: %v", err),
|
||||||
|
})
|
||||||
|
cloudLogger.Errorf("failed to initialize OIDC provider: %v", err)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -190,6 +217,7 @@ func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSess
|
||||||
|
|
||||||
googleIdentity := idToken.Audience[0] + ":" + idToken.Subject
|
googleIdentity := idToken.Audience[0] + ":" + idToken.Subject
|
||||||
if config.GoogleIdentity != googleIdentity {
|
if config.GoogleIdentity != googleIdentity {
|
||||||
|
_ = wsjson.Write(context.Background(), c, gin.H{"error": "google identity mismatch"})
|
||||||
return fmt.Errorf("google identity mismatch")
|
return fmt.Errorf("google identity mismatch")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -216,6 +244,9 @@ func handleSessionRequest(ctx context.Context, c *websocket.Conn, req WebRTCSess
|
||||||
_ = peerConn.Close()
|
_ = peerConn.Close()
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
cloudLogger.Info("new session accepted")
|
||||||
|
cloudLogger.Tracef("new session accepted: %v", session)
|
||||||
currentSession = session
|
currentSession = session
|
||||||
_ = wsjson.Write(context.Background(), c, gin.H{"sd": sd})
|
_ = wsjson.Write(context.Background(), c, gin.H{"sd": sd})
|
||||||
return nil
|
return nil
|
||||||
|
@ -225,7 +256,7 @@ func RunWebsocketClient() {
|
||||||
for {
|
for {
|
||||||
err := runWebsocketClient()
|
err := runWebsocketClient()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println("Websocket client error:", err)
|
cloudLogger.Errorf("websocket client error: %v", err)
|
||||||
time.Sleep(5 * time.Second)
|
time.Sleep(5 * time.Second)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -254,7 +285,7 @@ func rpcDeregisterDevice() error {
|
||||||
}
|
}
|
||||||
|
|
||||||
req.Header.Set("Authorization", "Bearer "+config.CloudToken)
|
req.Header.Set("Authorization", "Bearer "+config.CloudToken)
|
||||||
client := &http.Client{Timeout: 10 * time.Second}
|
client := &http.Client{Timeout: CloudAPIRequestTimeout}
|
||||||
resp, err := client.Do(req)
|
resp, err := client.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to send deregister request: %w", err)
|
return fmt.Errorf("failed to send deregister request: %w", err)
|
||||||
|
|
|
@ -47,6 +47,9 @@ var (
|
||||||
)
|
)
|
||||||
|
|
||||||
func LoadConfig() {
|
func LoadConfig() {
|
||||||
|
configLock.Lock()
|
||||||
|
defer configLock.Unlock()
|
||||||
|
|
||||||
if config != nil {
|
if config != nil {
|
||||||
logger.Info("config already loaded, skipping")
|
logger.Info("config already loaded, skipping")
|
||||||
return
|
return
|
||||||
|
|
|
@ -58,6 +58,9 @@ make build_dev
|
||||||
# Change directory to the binary output directory
|
# Change directory to the binary output directory
|
||||||
cd bin
|
cd bin
|
||||||
|
|
||||||
|
# Kill any existing instances of the application
|
||||||
|
ssh "${REMOTE_USER}@${REMOTE_HOST}" "killall jetkvm_app_debug || true"
|
||||||
|
|
||||||
# Copy the binary to the remote host
|
# Copy the binary to the remote host
|
||||||
cat jetkvm_app | ssh "${REMOTE_USER}@${REMOTE_HOST}" "cat > $REMOTE_PATH/jetkvm_app_debug"
|
cat jetkvm_app | ssh "${REMOTE_USER}@${REMOTE_HOST}" "cat > $REMOTE_PATH/jetkvm_app_debug"
|
||||||
|
|
||||||
|
@ -79,8 +82,7 @@ cd "$REMOTE_PATH"
|
||||||
chmod +x jetkvm_app_debug
|
chmod +x jetkvm_app_debug
|
||||||
|
|
||||||
# Run the application in the background
|
# Run the application in the background
|
||||||
./jetkvm_app_debug
|
PION_LOG_TRACE=jetkvm,cloud ./jetkvm_app_debug
|
||||||
|
|
||||||
EOF
|
EOF
|
||||||
|
|
||||||
echo "Deployment complete."
|
echo "Deployment complete."
|
||||||
|
|
1
log.go
1
log.go
|
@ -6,3 +6,4 @@ import "github.com/pion/logging"
|
||||||
// ref: https://github.com/pion/webrtc/wiki/Debugging-WebRTC
|
// ref: https://github.com/pion/webrtc/wiki/Debugging-WebRTC
|
||||||
var logger = logging.NewDefaultLoggerFactory().NewLogger("jetkvm")
|
var logger = logging.NewDefaultLoggerFactory().NewLogger("jetkvm")
|
||||||
var usbLogger = logging.NewDefaultLoggerFactory().NewLogger("usb")
|
var usbLogger = logging.NewDefaultLoggerFactory().NewLogger("usb")
|
||||||
|
var cloudLogger = logging.NewDefaultLoggerFactory().NewLogger("cloud")
|
||||||
|
|
Loading…
Reference in New Issue