// Package handler provides HTTP handlers for various endpoints. package handler import ( "context" "fmt" "net/http" "sync" "modelRT/constants" "modelRT/logger" "modelRT/network" "github.com/gin-gonic/gin" "github.com/gofrs/uuid" ) var globalMonitorState *SharedMonitorState func init() { globalMonitorState = NewSharedMonitorState() } // RealTimeMonitorHandler define real time data monitor process API // @Summary 开始或结束实时数据监控 // @Description 根据用户输入的组件token,从 modelRT 服务中开始或结束对于实时数据的监控 // @Tags RealTime Component // @Accept json // @Produce json // @Router /data/realtime [get] func RealTimeMonitorHandler(c *gin.Context) { var request network.RealTimeQueryRequest var monitorAction string var monitorID string if err := c.ShouldBindJSON(&request); err != nil { logger.Error(c, "failed to unmarshal real time query request", "error", err) c.JSON(http.StatusOK, network.FailureResponse{ Code: http.StatusBadRequest, Msg: err.Error(), }) return } if request.Action == constants.MonitorStartAction && request.MonitorID == "" { monitorAction = request.Action id, err := uuid.NewV4() if err != nil { logger.Error(c, "failed to generate monitor id", "error", err) c.JSON(http.StatusOK, network.FailureResponse{ Code: http.StatusBadRequest, Msg: err.Error(), }) return } monitorID = id.String() } else if request.Action == constants.MonitorStartAction && request.MonitorID != "" { monitorAction = constants.MonitorAppendAction monitorID = request.MonitorID } else if request.Action == constants.MonitorStopAction && request.MonitorID != "" { monitorAction = request.Action monitorID = request.MonitorID } switch monitorAction { case constants.MonitorStartAction: err := globalMonitorState.CreateConfig(monitorID, request.Components) if err != nil { logger.Error(c, "create real time data monitor config failed", "error", err) c.JSON(http.StatusOK, network.FailureResponse{ Code: http.StatusBadRequest, Msg: err.Error(), }) return } c.JSON(http.StatusOK, network.SuccessResponse{ Code: http.StatusOK, Msg: "success", PayLoad: map[string]interface{}{ "monitor_id": monitorID, }, }) return case constants.MonitorStopAction: globalMonitorState.RemoveTargets(c, monitorID, request.Components) case constants.MonitorAppendAction: err := globalMonitorState.AppendTargets(monitorID, request.Components) if err != nil { logger.Error(c, "append target to real time data monitor config failed", "error", err) c.JSON(http.StatusOK, network.FailureResponse{ Code: http.StatusBadRequest, Msg: err.Error(), }) return } default: err := fmt.Errorf("%w: %s", constants.ErrUnsupportedAction, request.Action) logger.Error(c, "unsupported action of real time data monitor request", "error", err) c.JSON(http.StatusOK, network.FailureResponse{ Code: http.StatusBadRequest, Msg: err.Error(), }) return } } // RealTimeMonitorComponent define struct of real time monitor component type RealTimeMonitorComponent struct { targets []string } // RealTimeMonitorConfig define struct of real time monitor config type RealTimeMonitorConfig struct { noticeChan chan struct{} components map[string]*RealTimeMonitorComponent } // SharedMonitorState define struct of shared monitor state with mutex type SharedMonitorState struct { monitorMap map[string]*RealTimeMonitorConfig mutex sync.RWMutex } // NewSharedMonitorState define function to create new SharedMonitorState func NewSharedMonitorState() *SharedMonitorState { return &SharedMonitorState{ monitorMap: make(map[string]*RealTimeMonitorConfig), } } // CreateConfig define function to create config in SharedMonitorState func (s *SharedMonitorState) CreateConfig(monitorID string, components []network.RealTimeComponentItem) error { s.mutex.Lock() defer s.mutex.Unlock() if _, exist := s.monitorMap[monitorID]; exist { return fmt.Errorf("monitorID %s already exists. Use AppendTargets to modify existing config", monitorID) } config := &RealTimeMonitorConfig{ noticeChan: make(chan struct{}), components: make(map[string]*RealTimeMonitorComponent), } for _, compent := range components { config.components[compent.Interval] = &RealTimeMonitorComponent{ targets: compent.Targets, } } s.monitorMap[monitorID] = config return nil } // AppendTargets define function to append targets in SharedMonitorState func (s *SharedMonitorState) AppendTargets(monitorID string, components []network.RealTimeComponentItem) error { s.mutex.Lock() defer s.mutex.Unlock() config, exist := s.monitorMap[monitorID] if !exist { return fmt.Errorf("monitorID %s not found. Use CreateConfig to start a new config", monitorID) } for _, compent := range components { interval := compent.Interval comp, compExist := config.components[interval] if !compExist { comp = &RealTimeMonitorComponent{ targets: comp.targets, } config.components[interval] = comp } else { comp.targets = append(comp.targets, comp.targets...) } } config.noticeChan <- struct{}{} return nil } // UpsertTargets define function to upsert targets in SharedMonitorState func (s *SharedMonitorState) UpsertTargets(monitorID string, interval string, newTargets []string) (isNewMonitor bool, err error) { s.mutex.Lock() defer s.mutex.Unlock() config, exist := s.monitorMap[monitorID] if !exist { config = &RealTimeMonitorConfig{ noticeChan: make(chan struct{}), components: make(map[string]*RealTimeMonitorComponent), } config.components[interval] = &RealTimeMonitorComponent{ targets: newTargets, } s.monitorMap[monitorID] = config return true, nil } comp, compExist := config.components[interval] if !compExist { comp = &RealTimeMonitorComponent{ targets: newTargets, } config.components[interval] = comp } else { comp.targets = append(comp.targets, newTargets...) } return false, nil } // Get define function to get value from SharedMonitorState func (s *SharedMonitorState) Get(monitorID, interval string) ([]string, bool) { s.mutex.RLock() defer s.mutex.RUnlock() config, ok := s.monitorMap[monitorID] if !ok { return nil, false } component, ok := config.components[interval] if !ok { return nil, false } return component.targets, true } // RemoveTargets define function to remove targets in SharedMonitorState func (s *SharedMonitorState) RemoveTargets(ctx context.Context, monitorID string, components []network.RealTimeComponentItem) error { s.mutex.Lock() defer s.mutex.Unlock() config, exist := s.monitorMap[monitorID] if !exist { return fmt.Errorf("monitorID %s not found", monitorID) } for _, compent := range components { interval := compent.Interval comp, compExist := config.components[interval] if !compExist { logger.Error(ctx, fmt.Sprintf("component with interval %s not found under monitorID %s", interval, monitorID), "monitorID", monitorID, "interval", interval) continue } targetsToRemoveMap := make(map[string]struct{}) for _, target := range compent.Targets { targetsToRemoveMap[target] = struct{}{} } var newTargets []string for _, existingTarget := range comp.targets { if _, found := targetsToRemoveMap[existingTarget]; !found { newTargets = append(newTargets, existingTarget) } } comp.targets = newTargets if len(comp.targets) == 0 { delete(config.components, interval) } } if len(config.components) == 0 { delete(s.monitorMap, monitorID) } config.noticeChan <- struct{}{} return nil }