119 lines
3.3 KiB
Go
119 lines
3.3 KiB
Go
package brain_control
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"regexp"
|
|
"time"
|
|
|
|
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
|
|
)
|
|
|
|
const (
|
|
sourceConfigPath = "/machine/v1/source-config"
|
|
runtimeStatusPath = "/machine/v1/runtime-status"
|
|
)
|
|
|
|
var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
|
|
|
|
type SendResponse struct {
|
|
StatusCode int
|
|
Body []byte
|
|
CorrelationID string
|
|
}
|
|
type SendFunc func(authorization string, body []byte, correlationID string, timeout time.Duration) (SendResponse, error)
|
|
|
|
type ConfigSender struct {
|
|
Signer mi.Signer
|
|
Send SendFunc
|
|
Enabled bool
|
|
Timeout time.Duration
|
|
MaxAttempts int
|
|
Sleep func(time.Duration)
|
|
}
|
|
|
|
func (s ConfigSender) Publish(config SourceConfig, correlationID string) (SendResponse, error) {
|
|
if !s.Enabled {
|
|
return SendResponse{}, errors.New("CONNECTOR_DISABLED")
|
|
}
|
|
if s.Send == nil || !requestIDPattern.MatchString(correlationID) {
|
|
return SendResponse{}, errors.New("invalid connector configuration")
|
|
}
|
|
if s.Timeout <= 0 {
|
|
s.Timeout = 5 * time.Second
|
|
}
|
|
if s.MaxAttempts == 0 {
|
|
s.MaxAttempts = 4
|
|
}
|
|
if s.MaxAttempts < 1 {
|
|
return SendResponse{}, errors.New("invalid connector retry policy")
|
|
}
|
|
if s.Sleep == nil {
|
|
s.Sleep = time.Sleep
|
|
}
|
|
if err := ValidateSourceConfig(config); err != nil {
|
|
return SendResponse{}, err
|
|
}
|
|
body, err := MarshalSourceConfig(config)
|
|
if err != nil {
|
|
return SendResponse{}, err
|
|
}
|
|
var last error
|
|
for attempt := 0; attempt < s.MaxAttempts; attempt++ {
|
|
token, mintErr := s.Signer.Mint("yovision-brain", []string{"source-config:write"}, "POST", sourceConfigPath, body)
|
|
if mintErr != nil {
|
|
return SendResponse{}, mintErr
|
|
}
|
|
response, sendErr := s.Send("Bearer "+token, body, correlationID, s.Timeout)
|
|
if sendErr == nil && response.StatusCode >= 200 && response.StatusCode < 300 {
|
|
return response, nil
|
|
}
|
|
if sendErr == nil && response.StatusCode < 500 {
|
|
return SendResponse{}, fmt.Errorf("source config rejected: %d", response.StatusCode)
|
|
}
|
|
if sendErr != nil {
|
|
last = sendErr
|
|
} else {
|
|
last = fmt.Errorf("source config remote status: %d", response.StatusCode)
|
|
}
|
|
if attempt+1 < s.MaxAttempts {
|
|
delay := time.Second << attempt
|
|
if delay > 30*time.Second {
|
|
delay = 30 * time.Second
|
|
}
|
|
s.Sleep(delay)
|
|
}
|
|
}
|
|
return SendResponse{}, fmt.Errorf("source config delivery exhausted: %w", last)
|
|
}
|
|
|
|
type RuntimeStatusEndpoint struct {
|
|
Verifier mi.Verifier
|
|
Store ProjectionStore
|
|
Enabled bool
|
|
MaxBodyBytes int
|
|
}
|
|
|
|
func (e RuntimeStatusEndpoint) Receive(authorization string, body []byte, correlationID string, expected map[string]int64) (ProjectionView, error) {
|
|
if !e.Enabled {
|
|
return ProjectionView{}, errors.New("CONNECTOR_DISABLED")
|
|
}
|
|
if !requestIDPattern.MatchString(correlationID) {
|
|
return ProjectionView{}, errors.New("INVALID_CORRELATION_ID")
|
|
}
|
|
if e.MaxBodyBytes == 0 {
|
|
e.MaxBodyBytes = 10 * 1024 * 1024
|
|
}
|
|
if len(body) > e.MaxBodyBytes {
|
|
return ProjectionView{}, errors.New("REQUEST_TOO_LARGE")
|
|
}
|
|
token, err := mi.BearerToken(authorization)
|
|
if err != nil {
|
|
return ProjectionView{}, err
|
|
}
|
|
if _, err = e.Verifier.Verify(token, "yovision-sense", "runtime-status:write", "POST", runtimeStatusPath, body); err != nil {
|
|
return ProjectionView{}, err
|
|
}
|
|
return e.Store.Ingest(body, expected)
|
|
}
|