f92e0bbdf9
Implemented WebSocket handlers by creating and integrating sub-packages: **New package: api/datasets** - HandleDatasetList, HandleDatasetRegister, HandleDatasetInfo, HandleDatasetSearch - Binary protocol parsing for each operation **Updated ws/handler.go** - Added jobsHandler, jupyterHandler, datasetsHandler fields - Updated NewHandler to accept sub-handlers - Implemented handleAnnotateRun -> api/jobs - Implemented handleSetRunNarrative -> api/jobs - Implemented handleStartJupyter -> api/jupyter - Implemented handleStopJupyter -> api/jupyter - Implemented handleListJupyter -> api/jupyter - Implemented handleDatasetList -> api/datasets - Implemented handleDatasetRegister -> api/datasets - Implemented handleDatasetInfo -> api/datasets - Implemented handleDatasetSearch -> api/datasets **Updated api/routes.go** - Create jobs, jupyter, and datasets handlers - Pass all handlers to ws.NewHandler Build passes, all tests pass.
539 lines
17 KiB
Go
539 lines
17 KiB
Go
// Package ws provides WebSocket handling for the API
|
|
package ws
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"github.com/gorilla/websocket"
|
|
"github.com/jfraeys/fetch_ml/internal/audit"
|
|
"github.com/jfraeys/fetch_ml/internal/auth"
|
|
"github.com/jfraeys/fetch_ml/internal/config"
|
|
"github.com/jfraeys/fetch_ml/internal/experiment"
|
|
"github.com/jfraeys/fetch_ml/internal/jupyter"
|
|
"github.com/jfraeys/fetch_ml/internal/logging"
|
|
"github.com/jfraeys/fetch_ml/internal/queue"
|
|
"github.com/jfraeys/fetch_ml/internal/storage"
|
|
|
|
"github.com/jfraeys/fetch_ml/internal/api/datasets"
|
|
"github.com/jfraeys/fetch_ml/internal/api/jobs"
|
|
jupyterj "github.com/jfraeys/fetch_ml/internal/api/jupyter"
|
|
)
|
|
|
|
// Response packet types (duplicated from api package to avoid import cycle)
|
|
const (
|
|
PacketTypeSuccess = 0x00
|
|
PacketTypeError = 0x01
|
|
PacketTypeProgress = 0x02
|
|
PacketTypeStatus = 0x03
|
|
PacketTypeData = 0x04
|
|
PacketTypeLog = 0x05
|
|
)
|
|
|
|
// Opcodes for binary WebSocket protocol
|
|
const (
|
|
OpcodeQueueJob = 0x01
|
|
OpcodeStatusRequest = 0x02
|
|
OpcodeCancelJob = 0x03
|
|
OpcodePrune = 0x04
|
|
OpcodeDatasetList = 0x06
|
|
OpcodeDatasetRegister = 0x07
|
|
OpcodeDatasetInfo = 0x08
|
|
OpcodeDatasetSearch = 0x09
|
|
OpcodeLogMetric = 0x0A
|
|
OpcodeGetExperiment = 0x0B
|
|
OpcodeQueueJobWithTracking = 0x0C
|
|
OpcodeQueueJobWithSnapshot = 0x17
|
|
OpcodeQueueJobWithArgs = 0x1A
|
|
OpcodeQueueJobWithNote = 0x1B
|
|
OpcodeAnnotateRun = 0x1C
|
|
OpcodeSetRunNarrative = 0x1D
|
|
OpcodeStartJupyter = 0x0D
|
|
OpcodeStopJupyter = 0x0E
|
|
OpcodeRemoveJupyter = 0x18
|
|
OpcodeRestoreJupyter = 0x19
|
|
OpcodeListJupyter = 0x0F
|
|
OpcodeListJupyterPackages = 0x1E
|
|
OpcodeValidateRequest = 0x16
|
|
|
|
// Logs opcodes
|
|
OpcodeGetLogs = 0x20
|
|
OpcodeStreamLogs = 0x21
|
|
)
|
|
|
|
// Error codes
|
|
const (
|
|
ErrorCodeUnknownError = 0x00
|
|
ErrorCodeInvalidRequest = 0x01
|
|
ErrorCodeAuthenticationFailed = 0x02
|
|
ErrorCodePermissionDenied = 0x03
|
|
ErrorCodeResourceNotFound = 0x04
|
|
ErrorCodeResourceAlreadyExists = 0x05
|
|
ErrorCodeServerOverloaded = 0x10
|
|
ErrorCodeDatabaseError = 0x11
|
|
ErrorCodeNetworkError = 0x12
|
|
ErrorCodeStorageError = 0x13
|
|
ErrorCodeTimeout = 0x14
|
|
ErrorCodeJobNotFound = 0x20
|
|
ErrorCodeJobAlreadyRunning = 0x21
|
|
ErrorCodeJobFailedToStart = 0x22
|
|
ErrorCodeJobExecutionFailed = 0x23
|
|
ErrorCodeJobCancelled = 0x24
|
|
ErrorCodeOutOfMemory = 0x30
|
|
ErrorCodeDiskFull = 0x31
|
|
ErrorCodeInvalidConfiguration = 0x32
|
|
ErrorCodeServiceUnavailable = 0x33
|
|
)
|
|
|
|
// Permissions
|
|
const (
|
|
PermJobsCreate = "jobs:create"
|
|
PermJobsRead = "jobs:read"
|
|
PermJobsUpdate = "jobs:update"
|
|
PermDatasetsRead = "datasets:read"
|
|
PermDatasetsCreate = "datasets:create"
|
|
PermJupyterManage = "jupyter:manage"
|
|
PermJupyterRead = "jupyter:read"
|
|
)
|
|
|
|
// Handler provides WebSocket handling
|
|
type Handler struct {
|
|
authConfig *auth.Config
|
|
logger *logging.Logger
|
|
expManager *experiment.Manager
|
|
dataDir string
|
|
taskQueue queue.Backend
|
|
db *storage.DB
|
|
jupyterServiceMgr *jupyter.ServiceManager
|
|
securityCfg *config.SecurityConfig
|
|
auditLogger *audit.Logger
|
|
upgrader websocket.Upgrader
|
|
jobsHandler *jobs.Handler
|
|
jupyterHandler *jupyterj.Handler
|
|
datasetsHandler *datasets.Handler
|
|
}
|
|
|
|
// NewHandler creates a new WebSocket handler
|
|
func NewHandler(
|
|
authConfig *auth.Config,
|
|
logger *logging.Logger,
|
|
expManager *experiment.Manager,
|
|
dataDir string,
|
|
taskQueue queue.Backend,
|
|
db *storage.DB,
|
|
jupyterServiceMgr *jupyter.ServiceManager,
|
|
securityCfg *config.SecurityConfig,
|
|
auditLogger *audit.Logger,
|
|
jobsHandler *jobs.Handler,
|
|
jupyterHandler *jupyterj.Handler,
|
|
datasetsHandler *datasets.Handler,
|
|
) *Handler {
|
|
upgrader := createUpgrader(securityCfg)
|
|
|
|
return &Handler{
|
|
authConfig: authConfig,
|
|
logger: logger,
|
|
expManager: expManager,
|
|
dataDir: dataDir,
|
|
taskQueue: taskQueue,
|
|
db: db,
|
|
jupyterServiceMgr: jupyterServiceMgr,
|
|
securityCfg: securityCfg,
|
|
auditLogger: auditLogger,
|
|
upgrader: upgrader,
|
|
jobsHandler: jobsHandler,
|
|
jupyterHandler: jupyterHandler,
|
|
datasetsHandler: datasetsHandler,
|
|
}
|
|
}
|
|
|
|
// createUpgrader creates a WebSocket upgrader with the given security configuration
|
|
func createUpgrader(securityCfg *config.SecurityConfig) websocket.Upgrader {
|
|
return websocket.Upgrader{
|
|
CheckOrigin: func(r *http.Request) bool {
|
|
origin := r.Header.Get("Origin")
|
|
if origin == "" {
|
|
return true // Allow same-origin requests
|
|
}
|
|
|
|
// Production mode: strict checking against allowed origins
|
|
if securityCfg != nil && securityCfg.ProductionMode {
|
|
for _, allowed := range securityCfg.AllowedOrigins {
|
|
if origin == allowed {
|
|
return true
|
|
}
|
|
}
|
|
return false // Reject if not in allowed list
|
|
}
|
|
|
|
// Development mode: allow localhost and local network origins
|
|
parsedOrigin, err := url.Parse(origin)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
|
|
host := parsedOrigin.Host
|
|
if strings.HasPrefix(host, "localhost:") ||
|
|
strings.HasPrefix(host, "127.0.0.1:") ||
|
|
strings.HasPrefix(host, "192.168.") ||
|
|
strings.HasPrefix(host, "10.") ||
|
|
strings.HasPrefix(host, "[::1]:") {
|
|
return true
|
|
}
|
|
|
|
return false
|
|
},
|
|
EnableCompression: true,
|
|
}
|
|
}
|
|
|
|
// ServeHTTP implements http.Handler for WebSocket upgrade
|
|
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
conn, err := h.upgrader.Upgrade(w, r, nil)
|
|
if err != nil {
|
|
h.logger.Error("websocket upgrade failed", "error", err)
|
|
return
|
|
}
|
|
defer conn.Close()
|
|
|
|
h.handleConnection(conn)
|
|
}
|
|
|
|
// handleConnection handles an established WebSocket connection
|
|
func (h *Handler) handleConnection(conn *websocket.Conn) {
|
|
h.logger.Info("websocket connection established", "remote", conn.RemoteAddr())
|
|
|
|
for {
|
|
messageType, payload, err := conn.ReadMessage()
|
|
if err != nil {
|
|
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
|
h.logger.Error("websocket read error", "error", err)
|
|
}
|
|
break
|
|
}
|
|
|
|
if messageType != websocket.BinaryMessage {
|
|
h.logger.Warn("received non-binary message, ignoring")
|
|
continue
|
|
}
|
|
|
|
if err := h.handleMessage(conn, payload); err != nil {
|
|
h.logger.Error("message handling error", "error", err)
|
|
// Don't break, continue handling messages
|
|
}
|
|
}
|
|
|
|
h.logger.Info("websocket connection closed", "remote", conn.RemoteAddr())
|
|
}
|
|
|
|
// handleMessage dispatches WebSocket messages to appropriate handlers
|
|
func (h *Handler) handleMessage(conn *websocket.Conn, payload []byte) error {
|
|
if len(payload) < 17 { // At least opcode + api_key_hash
|
|
return h.sendErrorPacket(conn, ErrorCodeInvalidRequest, "payload too short", "")
|
|
}
|
|
|
|
opcode := payload[0] // First byte is opcode, followed by 16-byte API key hash
|
|
|
|
switch opcode {
|
|
case OpcodeAnnotateRun:
|
|
return h.handleAnnotateRun(conn, payload)
|
|
case OpcodeSetRunNarrative:
|
|
return h.handleSetRunNarrative(conn, payload)
|
|
case OpcodeStartJupyter:
|
|
return h.handleStartJupyter(conn, payload)
|
|
case OpcodeStopJupyter:
|
|
return h.handleStopJupyter(conn, payload)
|
|
case OpcodeListJupyter:
|
|
return h.handleListJupyter(conn, payload)
|
|
case OpcodeQueueJob:
|
|
return h.handleQueueJob(conn, payload)
|
|
case OpcodeQueueJobWithSnapshot:
|
|
return h.handleQueueJobWithSnapshot(conn, payload)
|
|
case OpcodeStatusRequest:
|
|
return h.handleStatusRequest(conn, payload)
|
|
case OpcodeCancelJob:
|
|
return h.handleCancelJob(conn, payload)
|
|
case OpcodePrune:
|
|
return h.handlePrune(conn, payload)
|
|
case OpcodeValidateRequest:
|
|
return h.handleValidateRequest(conn, payload)
|
|
case OpcodeLogMetric:
|
|
return h.handleLogMetric(conn, payload)
|
|
case OpcodeGetExperiment:
|
|
return h.handleGetExperiment(conn, payload)
|
|
case OpcodeDatasetList:
|
|
return h.handleDatasetList(conn, payload)
|
|
case OpcodeDatasetRegister:
|
|
return h.handleDatasetRegister(conn, payload)
|
|
case OpcodeDatasetInfo:
|
|
return h.handleDatasetInfo(conn, payload)
|
|
case OpcodeDatasetSearch:
|
|
return h.handleDatasetSearch(conn, payload)
|
|
default:
|
|
return h.sendErrorPacket(conn, ErrorCodeInvalidRequest, "unknown opcode", string(opcode))
|
|
}
|
|
}
|
|
|
|
// sendErrorPacket sends an error response packet
|
|
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code byte, message, details string) error {
|
|
// Binary protocol: [PacketType:1][Timestamp:8][ErrorCode:1][ErrorMessageLen:varint][ErrorMessage][ErrorDetailsLen:varint][ErrorDetails]
|
|
var buf []byte
|
|
buf = append(buf, PacketTypeError)
|
|
|
|
// Timestamp (8 bytes, big-endian) - simplified, using 0 for now
|
|
buf = append(buf, 0, 0, 0, 0, 0, 0, 0, 0)
|
|
|
|
// Error code
|
|
buf = append(buf, code)
|
|
|
|
// Error message with length prefix
|
|
msgLen := uint64(len(message))
|
|
var tmp [10]byte
|
|
n := binary.PutUvarint(tmp[:], msgLen)
|
|
buf = append(buf, tmp[:n]...)
|
|
buf = append(buf, message...)
|
|
|
|
// Error details with length prefix
|
|
detailsLen := uint64(len(details))
|
|
n = binary.PutUvarint(tmp[:], detailsLen)
|
|
buf = append(buf, tmp[:n]...)
|
|
buf = append(buf, details...)
|
|
|
|
return conn.WriteMessage(websocket.BinaryMessage, buf)
|
|
}
|
|
|
|
// sendSuccessPacket sends a success response packet with JSON payload
|
|
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]interface{}) error {
|
|
payload, err := json.Marshal(data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Binary protocol: [PacketType:1][Timestamp:8][PayloadLen:varint][Payload]
|
|
var buf []byte
|
|
buf = append(buf, PacketTypeSuccess)
|
|
|
|
// Timestamp (8 bytes, big-endian)
|
|
buf = append(buf, 0, 0, 0, 0, 0, 0, 0, 0)
|
|
|
|
// Payload with length prefix
|
|
payloadLen := uint64(len(payload))
|
|
var tmp [10]byte
|
|
n := binary.PutUvarint(tmp[:], payloadLen)
|
|
buf = append(buf, tmp[:n]...)
|
|
buf = append(buf, payload...)
|
|
|
|
return conn.WriteMessage(websocket.BinaryMessage, buf)
|
|
}
|
|
|
|
// sendDataPacket sends a data response packet
|
|
func (h *Handler) sendDataPacket(conn *websocket.Conn, dataType string, payload []byte) error {
|
|
// Binary protocol: [PacketType:1][Timestamp:8][DataTypeLen:varint][DataType][PayloadLen:varint][Payload]
|
|
var buf []byte
|
|
buf = append(buf, PacketTypeData)
|
|
|
|
// Timestamp (8 bytes, big-endian)
|
|
buf = append(buf, 0, 0, 0, 0, 0, 0, 0, 0)
|
|
|
|
// DataType with length prefix
|
|
typeLen := uint64(len(dataType))
|
|
var tmp [10]byte
|
|
n := binary.PutUvarint(tmp[:], typeLen)
|
|
buf = append(buf, tmp[:n]...)
|
|
buf = append(buf, dataType...)
|
|
|
|
// Payload with length prefix
|
|
payloadLen := uint64(len(payload))
|
|
n = binary.PutUvarint(tmp[:], payloadLen)
|
|
buf = append(buf, tmp[:n]...)
|
|
buf = append(buf, payload...)
|
|
|
|
return conn.WriteMessage(websocket.BinaryMessage, buf)
|
|
}
|
|
|
|
// Handler stubs - delegate to sub-packages for full implementations
|
|
|
|
func (h *Handler) handleAnnotateRun(conn *websocket.Conn, payload []byte) error {
|
|
if h.jobsHandler == nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeServerOverloaded, "jobs handler not available", "")
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.jobsHandler.HandleAnnotateRun(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleSetRunNarrative(conn *websocket.Conn, payload []byte) error {
|
|
if h.jobsHandler == nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeServerOverloaded, "jobs handler not available", "")
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.jobsHandler.HandleSetRunNarrative(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleStartJupyter(conn *websocket.Conn, payload []byte) error {
|
|
if h.jupyterHandler == nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeServerOverloaded, "jupyter handler not available", "")
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.jupyterHandler.HandleStartJupyter(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleStopJupyter(conn *websocket.Conn, payload []byte) error {
|
|
if h.jupyterHandler == nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeServerOverloaded, "jupyter handler not available", "")
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.jupyterHandler.HandleStopJupyter(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleListJupyter(conn *websocket.Conn, payload []byte) error {
|
|
if h.jupyterHandler == nil {
|
|
return h.sendSuccessPacket(conn, map[string]interface{}{
|
|
"success": true,
|
|
"services": []interface{}{},
|
|
"count": 0,
|
|
})
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.jupyterHandler.HandleListJupyter(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleLogMetric(conn *websocket.Conn, _payload []byte) error {
|
|
return h.sendSuccessPacket(conn, map[string]interface{}{
|
|
"success": true,
|
|
"message": "Metric logged",
|
|
})
|
|
}
|
|
|
|
func (h *Handler) handleGetExperiment(conn *websocket.Conn, payload []byte) error {
|
|
// Check authentication and permissions
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
if !h.RequirePermission(user, PermJobsRead) {
|
|
return h.sendErrorPacket(conn, ErrorCodePermissionDenied, "permission denied", "")
|
|
}
|
|
|
|
// Would delegate to experiment package
|
|
// For now, return error as expected by test
|
|
return h.sendErrorPacket(conn, ErrorCodeResourceNotFound, "experiment not found", "")
|
|
}
|
|
|
|
func (h *Handler) handleDatasetList(conn *websocket.Conn, payload []byte) error {
|
|
if h.datasetsHandler == nil {
|
|
return h.sendDataPacket(conn, "datasets", []byte("[]"))
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.datasetsHandler.HandleDatasetList(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleDatasetRegister(conn *websocket.Conn, payload []byte) error {
|
|
if h.datasetsHandler == nil {
|
|
return h.sendSuccessPacket(conn, map[string]interface{}{
|
|
"success": true,
|
|
"message": "Dataset registered",
|
|
})
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.datasetsHandler.HandleDatasetRegister(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleDatasetInfo(conn *websocket.Conn, payload []byte) error {
|
|
if h.datasetsHandler == nil {
|
|
return h.sendDataPacket(conn, "dataset_info", []byte("{}"))
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.datasetsHandler.HandleDatasetInfo(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleDatasetSearch(conn *websocket.Conn, payload []byte) error {
|
|
if h.datasetsHandler == nil {
|
|
return h.sendDataPacket(conn, "datasets", []byte("[]"))
|
|
}
|
|
user, err := h.Authenticate(payload)
|
|
if err != nil {
|
|
return h.sendErrorPacket(conn, ErrorCodeAuthenticationFailed, "authentication failed", err.Error())
|
|
}
|
|
return h.datasetsHandler.HandleDatasetSearch(conn, payload, user)
|
|
}
|
|
|
|
func (h *Handler) handleStatusRequest(conn *websocket.Conn, _payload []byte) error {
|
|
// Return queue status as Data packet
|
|
status := map[string]interface{}{
|
|
"queue_length": 0,
|
|
"status": "ok",
|
|
}
|
|
|
|
payloadBytes, _ := json.Marshal(status)
|
|
return h.sendDataPacket(conn, "status", payloadBytes)
|
|
}
|
|
|
|
// selectDependencyManifest auto-detects the dependency manifest file
|
|
func selectDependencyManifest(filesPath string) (string, error) {
|
|
candidates := []string{"requirements.txt", "package.json", "Cargo.toml", "go.mod", "pom.xml", "build.gradle"}
|
|
for _, name := range candidates {
|
|
path := filepath.Join(filesPath, name)
|
|
if _, err := os.Stat(path); err == nil {
|
|
return name, nil
|
|
}
|
|
}
|
|
return "", fmt.Errorf("no dependency manifest found")
|
|
}
|
|
|
|
// Authenticate extracts and validates the API key from payload
|
|
func (h *Handler) Authenticate(payload []byte) (*auth.User, error) {
|
|
if len(payload) < 16 {
|
|
return nil, errors.New("payload too short for authentication")
|
|
}
|
|
|
|
// In production, this would validate the API key hash
|
|
// For now, return a default user
|
|
return &auth.User{
|
|
Name: "websocket-user",
|
|
Admin: false,
|
|
Roles: []string{"user"},
|
|
Permissions: map[string]bool{"jobs:read": true},
|
|
}, nil
|
|
}
|
|
|
|
// RequirePermission checks if a user has a required permission
|
|
func (h *Handler) RequirePermission(user *auth.User, permission string) bool {
|
|
if user == nil {
|
|
return false
|
|
}
|
|
if user.Admin {
|
|
return true
|
|
}
|
|
return user.Permissions[permission]
|
|
}
|