Compare commits
No commits in common. "feat/bootstrap-data-objects-from-pg-to-redis" and "develop" have entirely different histories.
feat/boots
...
develop
|
|
@ -30,7 +30,6 @@ go.work
|
||||||
/configs/**/*.pem
|
/configs/**/*.pem
|
||||||
|
|
||||||
# ai config
|
# ai config
|
||||||
.agents/
|
|
||||||
.cursor/
|
.cursor/
|
||||||
.claude/
|
.claude/
|
||||||
.codewhale/
|
.codewhale/
|
||||||
|
|
@ -42,6 +41,3 @@ go.work
|
||||||
ai-debug.log
|
ai-debug.log
|
||||||
*.patch
|
*.patch
|
||||||
*.diff
|
*.diff
|
||||||
docs/agents/
|
|
||||||
AGENTS.md
|
|
||||||
skills-lock.json
|
|
||||||
|
|
|
||||||
|
|
@ -1,255 +0,0 @@
|
||||||
// Package manualsync synchronizes measurement manual-mode changes with the
|
|
||||||
// protocol service responsible for the measurement's data source
|
|
||||||
package manualsync
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"math"
|
|
||||||
"net/http"
|
|
||||||
"net/url"
|
|
||||||
"strconv"
|
|
||||||
"strings"
|
|
||||||
|
|
||||||
"modelRT/config"
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/orm"
|
|
||||||
)
|
|
||||||
|
|
||||||
const maxErrorResponseBody = 4 << 10
|
|
||||||
|
|
||||||
// SyntheticData is one manually supplied measurement value
|
|
||||||
type SyntheticData struct {
|
|
||||||
Time int64 `json:"time"`
|
|
||||||
Value float64 `json:"value"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// Target identifies a measurement in a downstream protocol service
|
|
||||||
type Target struct {
|
|
||||||
Type int `json:"type"`
|
|
||||||
Station string `json:"station"`
|
|
||||||
MainPos string `json:"main_pos"`
|
|
||||||
SubPos string `json:"sub_pos"`
|
|
||||||
Option string `json:"option"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// Request is the payload accepted by POST /api/manual
|
|
||||||
type Request struct {
|
|
||||||
Mode int16 `json:"mode"`
|
|
||||||
Data []SyntheticData `json:"data,omitempty"`
|
|
||||||
Target Target `json:"target"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// Syncer synchronizes a measurement mode or manual-value change
|
|
||||||
type Syncer interface {
|
|
||||||
Sync(context.Context, orm.JSONMap, int16, *SyntheticData) error
|
|
||||||
}
|
|
||||||
|
|
||||||
// Client calls the protocol-specific manual synchronization endpoint
|
|
||||||
type Client struct {
|
|
||||||
httpClient *http.Client
|
|
||||||
protocolCL3611URL string
|
|
||||||
protocol104URL string
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewClient validates the configuration and constructs a reusable client
|
|
||||||
func NewClient(cfg config.ManualSyncConfig) (*Client, error) {
|
|
||||||
if cfg.Timeout <= 0 {
|
|
||||||
return nil, fmt.Errorf("manual sync timeout must be greater than zero")
|
|
||||||
}
|
|
||||||
protocolCL3611URL, err := endpointURL(cfg.ProtocolCL3611URL, cfg.APIPath)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("invalid protocol CL3611 URL: %w", err)
|
|
||||||
}
|
|
||||||
protocol104URL, err := endpointURL(cfg.Protocol104URL, cfg.APIPath)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("invalid protocol 104 URL: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return &Client{
|
|
||||||
httpClient: &http.Client{Timeout: cfg.Timeout},
|
|
||||||
protocolCL3611URL: protocolCL3611URL,
|
|
||||||
protocol104URL: protocol104URL,
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sync posts one mode transition or manual-value update. Data is omitted for
|
|
||||||
// mode transitions and included only when sample is non-nil in manual mode
|
|
||||||
func (c *Client) Sync(ctx context.Context, dataSource orm.JSONMap, mode int16, data *SyntheticData) error {
|
|
||||||
if c == nil || c.httpClient == nil {
|
|
||||||
return fmt.Errorf("manual sync client is not initialized")
|
|
||||||
}
|
|
||||||
if mode != constants.MeasurementModeManual && mode != constants.MeasurementModeAutomatic {
|
|
||||||
return fmt.Errorf("manual sync mode must be 0 or 1, got %d", mode)
|
|
||||||
}
|
|
||||||
if mode == constants.MeasurementModeAutomatic && data != nil {
|
|
||||||
return fmt.Errorf("automatic mode manual sync request cannot contain data")
|
|
||||||
}
|
|
||||||
|
|
||||||
endpoint, target, err := c.resolveTarget(dataSource)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
requestPayload := Request{Mode: mode, Target: target}
|
|
||||||
if data != nil {
|
|
||||||
requestPayload.Data = []SyntheticData{*data}
|
|
||||||
}
|
|
||||||
body, err := json.Marshal(requestPayload)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("encode manual sync request: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("create manual sync request: %w", err)
|
|
||||||
}
|
|
||||||
request.Header.Set("Content-Type", "application/json")
|
|
||||||
|
|
||||||
response, err := c.httpClient.Do(request)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("call manual sync endpoint: %w", err)
|
|
||||||
}
|
|
||||||
defer response.Body.Close()
|
|
||||||
if response.StatusCode >= http.StatusOK && response.StatusCode < http.StatusMultipleChoices {
|
|
||||||
_, _ = io.Copy(io.Discard, response.Body)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
responseBody, readErr := io.ReadAll(io.LimitReader(response.Body, maxErrorResponseBody))
|
|
||||||
if readErr != nil {
|
|
||||||
return fmt.Errorf("manual sync endpoint returned %s and response body could not be read: %w", response.Status, readErr)
|
|
||||||
}
|
|
||||||
message := strings.TrimSpace(string(responseBody))
|
|
||||||
if message == "" {
|
|
||||||
return fmt.Errorf("manual sync endpoint returned %s", response.Status)
|
|
||||||
}
|
|
||||||
return fmt.Errorf("manual sync endpoint returned %s: %s", response.Status, message)
|
|
||||||
}
|
|
||||||
|
|
||||||
type rawDataSource struct {
|
|
||||||
Type int `json:"type"`
|
|
||||||
IOAddress rawIOAddress `json:"io_address"`
|
|
||||||
}
|
|
||||||
|
|
||||||
type rawIOAddress struct {
|
|
||||||
DType int `json:"dtype"`
|
|
||||||
Station string `json:"station"`
|
|
||||||
Device string `json:"device"`
|
|
||||||
Channel string `json:"channel"`
|
|
||||||
Option string `json:"option"`
|
|
||||||
Packet any `json:"packet"`
|
|
||||||
Offset any `json:"offset"`
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *Client) resolveTarget(dataSource orm.JSONMap) (string, Target, error) {
|
|
||||||
if dataSource == nil {
|
|
||||||
return "", Target{}, fmt.Errorf("measurement data_source is null")
|
|
||||||
}
|
|
||||||
encoded, err := json.Marshal(dataSource)
|
|
||||||
if err != nil {
|
|
||||||
return "", Target{}, fmt.Errorf("encode measurement data_source: %w", err)
|
|
||||||
}
|
|
||||||
var source rawDataSource
|
|
||||||
if err := json.Unmarshal(encoded, &source); err != nil {
|
|
||||||
return "", Target{}, fmt.Errorf("decode measurement data_source: %w", err)
|
|
||||||
}
|
|
||||||
station := strings.TrimSpace(source.IOAddress.Station)
|
|
||||||
if station == "" {
|
|
||||||
return "", Target{}, fmt.Errorf("measurement data_source io_address.station is required")
|
|
||||||
}
|
|
||||||
|
|
||||||
switch source.Type {
|
|
||||||
case 1:
|
|
||||||
device := strings.TrimSpace(source.IOAddress.Device)
|
|
||||||
channel := strings.TrimSpace(source.IOAddress.Channel)
|
|
||||||
if device == "" {
|
|
||||||
return "", Target{}, fmt.Errorf("CL3611 data_source io_address.device is required")
|
|
||||||
}
|
|
||||||
if channel == "" {
|
|
||||||
return "", Target{}, fmt.Errorf("CL3611 data_source io_address.channel is required")
|
|
||||||
}
|
|
||||||
target := Target{
|
|
||||||
Station: station,
|
|
||||||
MainPos: device,
|
|
||||||
SubPos: channel,
|
|
||||||
}
|
|
||||||
switch source.IOAddress.DType {
|
|
||||||
case 1:
|
|
||||||
target.Type = 1
|
|
||||||
target.Option = strings.TrimSpace(source.IOAddress.Option)
|
|
||||||
case 2:
|
|
||||||
target.Type = 2
|
|
||||||
default:
|
|
||||||
return "", Target{}, fmt.Errorf("CL3611 data_source dtype must be 1 or 2, got %d", source.IOAddress.DType)
|
|
||||||
}
|
|
||||||
return c.protocolCL3611URL, target, nil
|
|
||||||
case 2:
|
|
||||||
packet, err := integerString(source.IOAddress.Packet)
|
|
||||||
if err != nil {
|
|
||||||
return "", Target{}, fmt.Errorf("104 data_source io_address.packet: %w", err)
|
|
||||||
}
|
|
||||||
offset, err := integerString(source.IOAddress.Offset)
|
|
||||||
if err != nil {
|
|
||||||
return "", Target{}, fmt.Errorf("104 data_source io_address.offset: %w", err)
|
|
||||||
}
|
|
||||||
return c.protocol104URL, Target{
|
|
||||||
Type: 3,
|
|
||||||
Station: station,
|
|
||||||
MainPos: packet,
|
|
||||||
SubPos: offset,
|
|
||||||
Option: "",
|
|
||||||
}, nil
|
|
||||||
default:
|
|
||||||
return "", Target{}, fmt.Errorf("unsupported measurement data_source type %d", source.Type)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func endpointURL(baseURL, apiPath string) (string, error) {
|
|
||||||
baseURL = strings.TrimSpace(baseURL)
|
|
||||||
if baseURL == "" {
|
|
||||||
return "", fmt.Errorf("base URL is required")
|
|
||||||
}
|
|
||||||
apiPath = strings.TrimSpace(apiPath)
|
|
||||||
if apiPath == "" {
|
|
||||||
return "", fmt.Errorf("API path is required")
|
|
||||||
}
|
|
||||||
endpoint := strings.TrimRight(baseURL, "/") + "/" + strings.TrimLeft(apiPath, "/")
|
|
||||||
parsed, err := url.Parse(endpoint)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
if parsed.Scheme != "http" && parsed.Scheme != "https" {
|
|
||||||
return "", fmt.Errorf("URL scheme must be http or https")
|
|
||||||
}
|
|
||||||
if parsed.Host == "" {
|
|
||||||
return "", fmt.Errorf("URL host is required")
|
|
||||||
}
|
|
||||||
return parsed.String(), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func integerString(value any) (string, error) {
|
|
||||||
switch typed := value.(type) {
|
|
||||||
case nil:
|
|
||||||
return "", fmt.Errorf("is required")
|
|
||||||
case float64:
|
|
||||||
if math.IsNaN(typed) || math.IsInf(typed, 0) || math.Trunc(typed) != typed {
|
|
||||||
return "", fmt.Errorf("must be an integer")
|
|
||||||
}
|
|
||||||
return strconv.FormatInt(int64(typed), 10), nil
|
|
||||||
case string:
|
|
||||||
trimmed := strings.TrimSpace(typed)
|
|
||||||
if trimmed == "" {
|
|
||||||
return "", fmt.Errorf("is required")
|
|
||||||
}
|
|
||||||
integer, err := strconv.ParseInt(trimmed, 10, 64)
|
|
||||||
if err != nil {
|
|
||||||
return "", fmt.Errorf("must be an integer: %w", err)
|
|
||||||
}
|
|
||||||
return strconv.FormatInt(integer, 10), nil
|
|
||||||
default:
|
|
||||||
return "", fmt.Errorf("has unsupported type %T", value)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,240 +0,0 @@
|
||||||
package manualsync
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"io"
|
|
||||||
"net/http"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"modelRT/config"
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/orm"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
type capturedRequest struct {
|
|
||||||
Path string
|
|
||||||
Method string
|
|
||||||
ContentType string
|
|
||||||
Payload Request
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientSyncRoutesAndMapsDataSources(t *testing.T) {
|
|
||||||
cl3611Requests := make(chan capturedRequest, 2)
|
|
||||||
protocol104Requests := make(chan capturedRequest, 1)
|
|
||||||
client, err := NewClient(config.ManualSyncConfig{
|
|
||||||
ProtocolCL3611URL: "http://cl3611.test",
|
|
||||||
Protocol104URL: "http://protocol104.test",
|
|
||||||
APIPath: "/api/manual",
|
|
||||||
Timeout: time.Second,
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
client.httpClient.Transport = roundTripFunc(func(request *http.Request) (*http.Response, error) {
|
|
||||||
var payload Request
|
|
||||||
require.NoError(t, json.NewDecoder(request.Body).Decode(&payload))
|
|
||||||
captured := capturedRequest{
|
|
||||||
Path: request.URL.Path,
|
|
||||||
Method: request.Method,
|
|
||||||
ContentType: request.Header.Get("Content-Type"),
|
|
||||||
Payload: payload,
|
|
||||||
}
|
|
||||||
if request.URL.Host == "cl3611.test" {
|
|
||||||
cl3611Requests <- captured
|
|
||||||
} else {
|
|
||||||
protocol104Requests <- captured
|
|
||||||
}
|
|
||||||
return httpResponse(http.StatusNoContent, ""), nil
|
|
||||||
})
|
|
||||||
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
dataSource orm.JSONMap
|
|
||||||
requests <-chan capturedRequest
|
|
||||||
wantTarget Target
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
name: "CL3611 phasor",
|
|
||||||
dataSource: orm.JSONMap{
|
|
||||||
"type": 1,
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"dtype": 1, "station": "001", "device": "ssu001",
|
|
||||||
"channel": "TM1", "option": "RMS",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
requests: cl3611Requests,
|
|
||||||
wantTarget: Target{
|
|
||||||
Type: 1, Station: "001", MainPos: "ssu001", SubPos: "TM1", Option: "RMS",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "CL3611 sample",
|
|
||||||
dataSource: orm.JSONMap{
|
|
||||||
"type": 1,
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"dtype": 2, "station": "002", "device": "ssu002",
|
|
||||||
"channel": "TS01", "option": "ignored",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
requests: cl3611Requests,
|
|
||||||
wantTarget: Target{
|
|
||||||
Type: 2, Station: "002", MainPos: "ssu002", SubPos: "TS01", Option: "",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "104",
|
|
||||||
dataSource: orm.JSONMap{
|
|
||||||
"type": 2,
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"station": "station000", "packet": 10, "offset": 35,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
requests: protocol104Requests,
|
|
||||||
wantTarget: Target{
|
|
||||||
Type: 3, Station: "station000", MainPos: "10", SubPos: "35", Option: "",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.name, func(t *testing.T) {
|
|
||||||
err := client.Sync(context.Background(), test.dataSource, constants.MeasurementModeAutomatic, nil)
|
|
||||||
require.NoError(t, err)
|
|
||||||
captured := <-test.requests
|
|
||||||
assert.Equal(t, http.MethodPost, captured.Method)
|
|
||||||
assert.Equal(t, "/api/manual", captured.Path)
|
|
||||||
assert.Equal(t, "application/json", captured.ContentType)
|
|
||||||
assert.Equal(t, constants.MeasurementModeAutomatic, captured.Payload.Mode)
|
|
||||||
assert.Nil(t, captured.Payload.Data)
|
|
||||||
assert.Equal(t, test.wantTarget, captured.Payload.Target)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientSyncIncludesManualValueData(t *testing.T) {
|
|
||||||
requests := make(chan capturedRequest, 1)
|
|
||||||
client, err := NewClient(config.ManualSyncConfig{
|
|
||||||
ProtocolCL3611URL: "http://cl3611.test",
|
|
||||||
Protocol104URL: "http://protocol104.test",
|
|
||||||
APIPath: "api/manual",
|
|
||||||
Timeout: time.Second,
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
client.httpClient.Transport = captureTransport(t, requests)
|
|
||||||
|
|
||||||
sample := SyntheticData{Time: 1736305467506000000, Value: 1.25}
|
|
||||||
err = client.Sync(context.Background(), orm.JSONMap{
|
|
||||||
"type": 2,
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"station": "station000", "packet": 1, "offset": 2,
|
|
||||||
},
|
|
||||||
}, constants.MeasurementModeManual, &sample)
|
|
||||||
require.NoError(t, err)
|
|
||||||
captured := <-requests
|
|
||||||
assert.Equal(t, constants.MeasurementModeManual, captured.Payload.Mode)
|
|
||||||
assert.Equal(t, []SyntheticData{sample}, captured.Payload.Data)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientSyncRejectsInvalidDataSources(t *testing.T) {
|
|
||||||
client, err := NewClient(config.ManualSyncConfig{
|
|
||||||
ProtocolCL3611URL: "http://cl3611.test",
|
|
||||||
Protocol104URL: "http://protocol104.test",
|
|
||||||
APIPath: "/api/manual",
|
|
||||||
Timeout: time.Second,
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
client.httpClient.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
||||||
t.Fatal("invalid data source must not call endpoint")
|
|
||||||
return nil, nil
|
|
||||||
})
|
|
||||||
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
dataSource orm.JSONMap
|
|
||||||
wantError string
|
|
||||||
}{
|
|
||||||
{name: "unsupported type", dataSource: orm.JSONMap{"type": 3, "io_address": map[string]any{"station": "s"}}, wantError: "unsupported"},
|
|
||||||
{name: "invalid dtype", dataSource: orm.JSONMap{"type": 1, "io_address": map[string]any{"dtype": 3, "station": "s", "device": "d", "channel": "c"}}, wantError: "dtype"},
|
|
||||||
{name: "missing station", dataSource: orm.JSONMap{"type": 2, "io_address": map[string]any{"packet": 1, "offset": 2}}, wantError: "station"},
|
|
||||||
{name: "fractional packet", dataSource: orm.JSONMap{"type": 2, "io_address": map[string]any{"station": "s", "packet": 1.5, "offset": 2}}, wantError: "packet"},
|
|
||||||
}
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.name, func(t *testing.T) {
|
|
||||||
err := client.Sync(context.Background(), test.dataSource, constants.MeasurementModeManual, nil)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), test.wantError)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientSyncReturnsNonSuccessResponse(t *testing.T) {
|
|
||||||
client, err := NewClient(config.ManualSyncConfig{
|
|
||||||
ProtocolCL3611URL: "http://cl3611.test",
|
|
||||||
Protocol104URL: "http://protocol104.test",
|
|
||||||
APIPath: "/api/manual",
|
|
||||||
Timeout: time.Second,
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
client.httpClient.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) {
|
|
||||||
return httpResponse(http.StatusServiceUnavailable, "downstream unavailable"), nil
|
|
||||||
})
|
|
||||||
|
|
||||||
err = client.Sync(context.Background(), orm.JSONMap{
|
|
||||||
"type": 2,
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"station": "s", "packet": 1, "offset": 2,
|
|
||||||
},
|
|
||||||
}, constants.MeasurementModeManual, nil)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "Service Unavailable")
|
|
||||||
assert.Contains(t, err.Error(), "downstream unavailable")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestNewClientValidatesConfiguration(t *testing.T) {
|
|
||||||
_, err := NewClient(config.ManualSyncConfig{})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "timeout")
|
|
||||||
|
|
||||||
_, err = NewClient(config.ManualSyncConfig{
|
|
||||||
ProtocolCL3611URL: "127.0.0.1:9001",
|
|
||||||
Protocol104URL: "http://127.0.0.1:9002",
|
|
||||||
APIPath: "/api/manual",
|
|
||||||
Timeout: time.Second,
|
|
||||||
})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "invalid protocol CL3611 URL")
|
|
||||||
}
|
|
||||||
|
|
||||||
type roundTripFunc func(*http.Request) (*http.Response, error)
|
|
||||||
|
|
||||||
func (roundTrip roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
|
|
||||||
return roundTrip(request)
|
|
||||||
}
|
|
||||||
|
|
||||||
func captureTransport(t *testing.T, requests chan<- capturedRequest) http.RoundTripper {
|
|
||||||
t.Helper()
|
|
||||||
return roundTripFunc(func(request *http.Request) (*http.Response, error) {
|
|
||||||
var payload Request
|
|
||||||
require.NoError(t, json.NewDecoder(request.Body).Decode(&payload))
|
|
||||||
requests <- capturedRequest{
|
|
||||||
Path: request.URL.Path,
|
|
||||||
Method: request.Method,
|
|
||||||
ContentType: request.Header.Get("Content-Type"),
|
|
||||||
Payload: payload,
|
|
||||||
}
|
|
||||||
return httpResponse(http.StatusNoContent, ""), nil
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
func httpResponse(status int, body string) *http.Response {
|
|
||||||
return &http.Response{
|
|
||||||
StatusCode: status,
|
|
||||||
Status: http.StatusText(status),
|
|
||||||
Body: io.NopCloser(strings.NewReader(body)),
|
|
||||||
Header: make(http.Header),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,33 +0,0 @@
|
||||||
package manualsync
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"sync"
|
|
||||||
|
|
||||||
"modelRT/orm"
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
defaultSyncerMu sync.RWMutex
|
|
||||||
defaultSyncer Syncer
|
|
||||||
)
|
|
||||||
|
|
||||||
// SetDefaultSyncer sets the process-wide manual measurement syncer.
|
|
||||||
// Passing nil clears the current default.
|
|
||||||
func SetDefaultSyncer(syncer Syncer) {
|
|
||||||
defaultSyncerMu.Lock()
|
|
||||||
defer defaultSyncerMu.Unlock()
|
|
||||||
defaultSyncer = syncer
|
|
||||||
}
|
|
||||||
|
|
||||||
// Sync uses the process-wide manual measurement syncer.
|
|
||||||
func Sync(ctx context.Context, dataSource orm.JSONMap, mode int16, sample *SyntheticData) error {
|
|
||||||
defaultSyncerMu.RLock()
|
|
||||||
syncer := defaultSyncer
|
|
||||||
defaultSyncerMu.RUnlock()
|
|
||||||
if syncer == nil {
|
|
||||||
return fmt.Errorf("manual measurement sync client is not initialized")
|
|
||||||
}
|
|
||||||
return syncer.Sync(ctx, dataSource, mode, sample)
|
|
||||||
}
|
|
||||||
|
|
@ -1,46 +0,0 @@
|
||||||
package manualsync
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/orm"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
type syncerFunc func(context.Context, orm.JSONMap, int16, *SyntheticData) error
|
|
||||||
|
|
||||||
func (syncer syncerFunc) Sync(ctx context.Context, dataSource orm.JSONMap, mode int16, sample *SyntheticData) error {
|
|
||||||
return syncer(ctx, dataSource, mode, sample)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestSyncRequiresDefaultSyncer(t *testing.T) {
|
|
||||||
SetDefaultSyncer(nil)
|
|
||||||
t.Cleanup(func() { SetDefaultSyncer(nil) })
|
|
||||||
|
|
||||||
err := Sync(context.Background(), nil, constants.MeasurementModeManual, nil)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "not initialized")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestSyncUsesDefaultSyncer(t *testing.T) {
|
|
||||||
SetDefaultSyncer(nil)
|
|
||||||
t.Cleanup(func() { SetDefaultSyncer(nil) })
|
|
||||||
|
|
||||||
dataSource := orm.JSONMap{"type": 2}
|
|
||||||
sample := &SyntheticData{Time: 123, Value: 4.5}
|
|
||||||
called := false
|
|
||||||
SetDefaultSyncer(syncerFunc(func(_ context.Context, gotDataSource orm.JSONMap, gotMode int16, gotSample *SyntheticData) error {
|
|
||||||
called = true
|
|
||||||
assert.Equal(t, dataSource, gotDataSource)
|
|
||||||
assert.Equal(t, constants.MeasurementModeManual, gotMode)
|
|
||||||
assert.Same(t, sample, gotSample)
|
|
||||||
return nil
|
|
||||||
}))
|
|
||||||
|
|
||||||
require.NoError(t, Sync(context.Background(), dataSource, constants.MeasurementModeManual, sample))
|
|
||||||
assert.True(t, called)
|
|
||||||
}
|
|
||||||
|
|
@ -1,22 +0,0 @@
|
||||||
package main
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"log/slog"
|
|
||||||
"net"
|
|
||||||
"os"
|
|
||||||
"os/signal"
|
|
||||||
"syscall"
|
|
||||||
)
|
|
||||||
|
|
||||||
func main() {
|
|
||||||
logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
|
|
||||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
||||||
|
|
||||||
err := run(ctx, logger, net.Listen)
|
|
||||||
stop()
|
|
||||||
if err != nil {
|
|
||||||
logger.Error("manual sync mock stopped", "error", err)
|
|
||||||
os.Exit(1)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,168 +0,0 @@
|
||||||
package main
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"log/slog"
|
|
||||||
"net"
|
|
||||||
"net/http"
|
|
||||||
"strconv"
|
|
||||||
"strings"
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"modelRT/client/manualsync"
|
|
||||||
)
|
|
||||||
|
|
||||||
type listenFunc func(network, address string) (net.Listener, error)
|
|
||||||
|
|
||||||
const (
|
|
||||||
cl3611Address = ":9001"
|
|
||||||
protocol104Address = ":9002"
|
|
||||||
manualAPIPath = "/api/manual"
|
|
||||||
healthPath = "/healthz"
|
|
||||||
maxRequestBody = 1 << 20
|
|
||||||
shutdownTimeout = 5 * time.Second
|
|
||||||
)
|
|
||||||
|
|
||||||
func newProtocolServer(protocol, address string, logger *slog.Logger) *http.Server {
|
|
||||||
mux := http.NewServeMux()
|
|
||||||
mux.HandleFunc("POST "+manualAPIPath, manualHandler(protocol, strings.TrimPrefix(address, ":"), logger))
|
|
||||||
mux.HandleFunc("GET "+healthPath, healthHandler)
|
|
||||||
|
|
||||||
return &http.Server{
|
|
||||||
Addr: address,
|
|
||||||
Handler: mux,
|
|
||||||
ReadHeaderTimeout: 5 * time.Second,
|
|
||||||
ReadTimeout: 10 * time.Second,
|
|
||||||
WriteTimeout: 10 * time.Second,
|
|
||||||
IdleTimeout: 60 * time.Second,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func run(ctx context.Context, logger *slog.Logger, listen listenFunc) error {
|
|
||||||
cl3611Listener, err := listen("tcp", cl3611Address)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("listen on %s: %w", cl3611Address, err)
|
|
||||||
}
|
|
||||||
protocol104Listener, err := listen("tcp", protocol104Address)
|
|
||||||
if err != nil {
|
|
||||||
_ = cl3611Listener.Close()
|
|
||||||
return fmt.Errorf("listen on %s: %w", protocol104Address, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
servers := []struct {
|
|
||||||
protocol string
|
|
||||||
server *http.Server
|
|
||||||
listener net.Listener
|
|
||||||
}{
|
|
||||||
{protocol: "cl3611", server: newProtocolServer("cl3611", cl3611Address, logger), listener: cl3611Listener},
|
|
||||||
{protocol: "104", server: newProtocolServer("104", protocol104Address, logger), listener: protocol104Listener},
|
|
||||||
}
|
|
||||||
|
|
||||||
serveErrors := make(chan error, len(servers))
|
|
||||||
var serversWaitGroup sync.WaitGroup
|
|
||||||
for _, configuredServer := range servers {
|
|
||||||
serversWaitGroup.Add(1)
|
|
||||||
go func() {
|
|
||||||
defer serversWaitGroup.Done()
|
|
||||||
logger.Info("manual sync mock server started",
|
|
||||||
"protocol", configuredServer.protocol,
|
|
||||||
"listen_addr", configuredServer.listener.Addr().String(),
|
|
||||||
)
|
|
||||||
if err := configuredServer.server.Serve(configuredServer.listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
||||||
serveErrors <- fmt.Errorf("%s server failed: %w", configuredServer.protocol, err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
var runError error
|
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
|
||||||
case runError = <-serveErrors:
|
|
||||||
}
|
|
||||||
|
|
||||||
shutdownContext, cancelShutdown := context.WithTimeout(context.Background(), shutdownTimeout)
|
|
||||||
defer cancelShutdown()
|
|
||||||
for _, configuredServer := range servers {
|
|
||||||
if err := configuredServer.server.Shutdown(shutdownContext); err != nil {
|
|
||||||
runError = errors.Join(runError, fmt.Errorf("shut down %s server: %w", configuredServer.protocol, err))
|
|
||||||
_ = configuredServer.server.Close()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
serversWaitGroup.Wait()
|
|
||||||
return runError
|
|
||||||
}
|
|
||||||
|
|
||||||
func healthHandler(response http.ResponseWriter, _ *http.Request) {
|
|
||||||
response.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
|
||||||
response.WriteHeader(http.StatusOK)
|
|
||||||
_, _ = response.Write([]byte("ok\n"))
|
|
||||||
}
|
|
||||||
|
|
||||||
func manualHandler(protocol, listenPort string, logger *slog.Logger) http.HandlerFunc {
|
|
||||||
return func(response http.ResponseWriter, request *http.Request) {
|
|
||||||
request.Body = http.MaxBytesReader(response, request.Body, maxRequestBody)
|
|
||||||
var payload manualsync.Request
|
|
||||||
decoder := json.NewDecoder(request.Body)
|
|
||||||
if err := decoder.Decode(&payload); err != nil {
|
|
||||||
logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
|
|
||||||
slog.String("protocol", protocol),
|
|
||||||
slog.String("listen_port", listenPort),
|
|
||||||
slog.String("remote_addr", request.RemoteAddr),
|
|
||||||
slog.Any("error", err),
|
|
||||||
)
|
|
||||||
var maxBytesError *http.MaxBytesError
|
|
||||||
if errors.As(err, &maxBytesError) {
|
|
||||||
http.Error(response, "request body too large", http.StatusRequestEntityTooLarge)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
http.Error(response, "invalid JSON request body", http.StatusBadRequest)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
var trailing any
|
|
||||||
if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
|
|
||||||
if err == nil {
|
|
||||||
err = errors.New("request body contains multiple JSON documents")
|
|
||||||
}
|
|
||||||
logger.LogAttrs(request.Context(), slog.LevelWarn, "invalid manual sync request",
|
|
||||||
slog.String("protocol", protocol),
|
|
||||||
slog.String("listen_port", listenPort),
|
|
||||||
slog.String("remote_addr", request.RemoteAddr),
|
|
||||||
slog.Any("error", err),
|
|
||||||
)
|
|
||||||
http.Error(response, "invalid JSON request body", http.StatusBadRequest)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.LogAttrs(request.Context(), slog.LevelInfo, "manual sync request received",
|
|
||||||
slog.String("protocol", protocol),
|
|
||||||
slog.String("listen_port", listenPort),
|
|
||||||
slog.String("remote_addr", request.RemoteAddr),
|
|
||||||
slog.Int("mode", int(payload.Mode)),
|
|
||||||
slog.Group("target",
|
|
||||||
slog.Int("type", payload.Target.Type),
|
|
||||||
slog.String("station", payload.Target.Station),
|
|
||||||
slog.String("main_pos", payload.Target.MainPos),
|
|
||||||
slog.String("sub_pos", payload.Target.SubPos),
|
|
||||||
slog.String("option", payload.Target.Option),
|
|
||||||
),
|
|
||||||
slog.Attr{Key: "data", Value: samplesLogValue(payload.Data)},
|
|
||||||
)
|
|
||||||
response.WriteHeader(http.StatusNoContent)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func samplesLogValue(samples []manualsync.SyntheticData) slog.Value {
|
|
||||||
attributes := make([]slog.Attr, 0, len(samples))
|
|
||||||
for index, sample := range samples {
|
|
||||||
attributes = append(attributes, slog.Group(strconv.Itoa(index),
|
|
||||||
slog.Int64("time", sample.Time),
|
|
||||||
slog.Float64("value", sample.Value),
|
|
||||||
))
|
|
||||||
}
|
|
||||||
return slog.GroupValue(attributes...)
|
|
||||||
}
|
|
||||||
|
|
@ -1,234 +0,0 @@
|
||||||
package main
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"io"
|
|
||||||
"log/slog"
|
|
||||||
"net"
|
|
||||||
"net/http"
|
|
||||||
"net/http/httptest"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestHealthEndpointReportsServerIsReady(t *testing.T) {
|
|
||||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
|
||||||
request := httptest.NewRequest(http.MethodGet, healthPath, nil)
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
result := response.Result()
|
|
||||||
defer result.Body.Close()
|
|
||||||
body, err := io.ReadAll(result.Body)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("read response body: %v", err)
|
|
||||||
}
|
|
||||||
if result.StatusCode != http.StatusOK {
|
|
||||||
t.Fatalf("status = %d, want %d", result.StatusCode, http.StatusOK)
|
|
||||||
}
|
|
||||||
if contentType := result.Header.Get("Content-Type"); contentType != "text/plain; charset=utf-8" {
|
|
||||||
t.Errorf("Content-Type = %q, want %q", contentType, "text/plain; charset=utf-8")
|
|
||||||
}
|
|
||||||
if string(body) != "ok\n" {
|
|
||||||
t.Errorf("body = %q, want %q", body, "ok\\n")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestManualEndpointLogsStructuredRequest(t *testing.T) {
|
|
||||||
var logs bytes.Buffer
|
|
||||||
logger := slog.New(slog.NewTextHandler(&logs, nil))
|
|
||||||
server := newProtocolServer("cl3611", cl3611Address, logger)
|
|
||||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{
|
|
||||||
"mode": 1,
|
|
||||||
"data": [{"time": 1736305467506000000, "value": 1.25}],
|
|
||||||
"target": {
|
|
||||||
"type": 1,
|
|
||||||
"station": "001",
|
|
||||||
"main_pos": "ssu001",
|
|
||||||
"sub_pos": "TM1",
|
|
||||||
"option": "RMS"
|
|
||||||
}
|
|
||||||
}`))
|
|
||||||
request.RemoteAddr = "127.0.0.1:52130"
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
if response.Code != http.StatusNoContent {
|
|
||||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusNoContent)
|
|
||||||
}
|
|
||||||
for _, fragment := range []string{
|
|
||||||
"protocol=cl3611",
|
|
||||||
"listen_port=9001",
|
|
||||||
"remote_addr=127.0.0.1:52130",
|
|
||||||
"mode=1",
|
|
||||||
"target.type=1",
|
|
||||||
"target.station=001",
|
|
||||||
"target.main_pos=ssu001",
|
|
||||||
"target.sub_pos=TM1",
|
|
||||||
"target.option=RMS",
|
|
||||||
"time=1736305467506000000",
|
|
||||||
"value=1.25",
|
|
||||||
} {
|
|
||||||
if !strings.Contains(logs.String(), fragment) {
|
|
||||||
t.Errorf("log %q does not contain %q", logs.String(), fragment)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestManualEndpointRejectsInvalidJSONAndLogsTheError(t *testing.T) {
|
|
||||||
var logs bytes.Buffer
|
|
||||||
logger := slog.New(slog.NewTextHandler(&logs, nil))
|
|
||||||
server := newProtocolServer("104", protocol104Address, logger)
|
|
||||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":`))
|
|
||||||
request.RemoteAddr = "127.0.0.1:52131"
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
if response.Code != http.StatusBadRequest {
|
|
||||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
|
|
||||||
}
|
|
||||||
for _, fragment := range []string{
|
|
||||||
"level=WARN",
|
|
||||||
"msg=\"invalid manual sync request\"",
|
|
||||||
"protocol=104",
|
|
||||||
"listen_port=9002",
|
|
||||||
"remote_addr=127.0.0.1:52131",
|
|
||||||
"error=",
|
|
||||||
} {
|
|
||||||
if !strings.Contains(logs.String(), fragment) {
|
|
||||||
t.Errorf("log %q does not contain %q", logs.String(), fragment)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestManualEndpointRejectsOversizedRequest(t *testing.T) {
|
|
||||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
|
||||||
body := `{"extra":"` + strings.Repeat("a", maxRequestBody) + `"}`
|
|
||||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(body))
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
if response.Code != http.StatusRequestEntityTooLarge {
|
|
||||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusRequestEntityTooLarge)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestManualEndpointRejectsMultipleJSONDocuments(t *testing.T) {
|
|
||||||
server := newProtocolServer("104", protocol104Address, slog.Default())
|
|
||||||
request := httptest.NewRequest(http.MethodPost, manualAPIPath, strings.NewReader(`{"mode":1}{"mode":0}`))
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
if response.Code != http.StatusBadRequest {
|
|
||||||
t.Fatalf("status = %d, want %d", response.Code, http.StatusBadRequest)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestRunClosesFirstListenerWhenSecondListenerFails(t *testing.T) {
|
|
||||||
rawListener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("listen on temporary port: %v", err)
|
|
||||||
}
|
|
||||||
firstListener := rawListener.(*net.TCPListener)
|
|
||||||
listenCalls := 0
|
|
||||||
listen := func(_, _ string) (net.Listener, error) {
|
|
||||||
listenCalls++
|
|
||||||
if listenCalls == 1 {
|
|
||||||
return firstListener, nil
|
|
||||||
}
|
|
||||||
return nil, errors.New("port is already in use")
|
|
||||||
}
|
|
||||||
|
|
||||||
err = run(context.Background(), slog.Default(), listen)
|
|
||||||
|
|
||||||
if err == nil || !strings.Contains(err.Error(), protocol104Address) {
|
|
||||||
t.Fatalf("run error = %v, want error containing %q", err, protocol104Address)
|
|
||||||
}
|
|
||||||
if err := firstListener.SetDeadline(time.Now()); err == nil {
|
|
||||||
t.Fatal("first listener is still open")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestRunServesBothProtocolsAndStopsOnCancellation(t *testing.T) {
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
addresses := make(chan string, 2)
|
|
||||||
listen := func(_, _ string) (net.Listener, error) {
|
|
||||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
||||||
if err == nil {
|
|
||||||
addresses <- listener.Addr().String()
|
|
||||||
}
|
|
||||||
return listener, err
|
|
||||||
}
|
|
||||||
runErrors := make(chan error, 1)
|
|
||||||
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
|
||||||
go func() {
|
|
||||||
runErrors <- run(ctx, logger, listen)
|
|
||||||
}()
|
|
||||||
|
|
||||||
cl3611Addr := <-addresses
|
|
||||||
protocol104Addr := <-addresses
|
|
||||||
waitForHealthyEndpoint(t, cl3611Addr)
|
|
||||||
waitForHealthyEndpoint(t, protocol104Addr)
|
|
||||||
cancel()
|
|
||||||
|
|
||||||
select {
|
|
||||||
case err := <-runErrors:
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("run returned an error during graceful shutdown: %v", err)
|
|
||||||
}
|
|
||||||
case <-time.After(2 * time.Second):
|
|
||||||
t.Fatal("servers did not stop after context cancellation")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestProtocolServerEnforcesMethodAndPathRouting(t *testing.T) {
|
|
||||||
server := newProtocolServer("cl3611", cl3611Address, slog.Default())
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
method string
|
|
||||||
path string
|
|
||||||
wantStatus int
|
|
||||||
}{
|
|
||||||
{name: "manual endpoint rejects GET", method: http.MethodGet, path: manualAPIPath, wantStatus: http.StatusMethodNotAllowed},
|
|
||||||
{name: "unknown path is not found", method: http.MethodGet, path: "/unknown", wantStatus: http.StatusNotFound},
|
|
||||||
}
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.name, func(t *testing.T) {
|
|
||||||
request := httptest.NewRequest(test.method, test.path, nil)
|
|
||||||
response := httptest.NewRecorder()
|
|
||||||
|
|
||||||
server.Handler.ServeHTTP(response, request)
|
|
||||||
|
|
||||||
if response.Code != test.wantStatus {
|
|
||||||
t.Fatalf("status = %d, want %d", response.Code, test.wantStatus)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func waitForHealthyEndpoint(t *testing.T, address string) {
|
|
||||||
t.Helper()
|
|
||||||
client := &http.Client{Timeout: 100 * time.Millisecond}
|
|
||||||
deadline := time.Now().Add(2 * time.Second)
|
|
||||||
for {
|
|
||||||
response, err := client.Get("http://" + address + healthPath)
|
|
||||||
if err == nil {
|
|
||||||
_ = response.Body.Close()
|
|
||||||
if response.StatusCode == http.StatusOK {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if time.Now().After(deadline) {
|
|
||||||
t.Fatalf("health endpoint at %s did not become ready", address)
|
|
||||||
}
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -91,18 +91,12 @@ type AntsConfig struct {
|
||||||
RTDReceiveConcurrentQuantity int `mapstructure:"rtd_receive_concurrent_quantity"` // polling real time data concurrent quantity
|
RTDReceiveConcurrentQuantity int `mapstructure:"rtd_receive_concurrent_quantity"` // polling real time data concurrent quantity
|
||||||
}
|
}
|
||||||
|
|
||||||
// ManualSyncConfig defines protocol endpoints used to synchronize manual
|
// DataRTConfig define config struct of data runtime server api config
|
||||||
// measurement mode and value changes.
|
|
||||||
type ManualSyncConfig struct {
|
|
||||||
ProtocolCL3611URL string `mapstructure:"protocol_cl3611_url"`
|
|
||||||
Protocol104URL string `mapstructure:"protocol_104_url"`
|
|
||||||
APIPath string `mapstructure:"api_path"`
|
|
||||||
Timeout time.Duration `mapstructure:"timeout"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// DataRTConfig defines APIs provided by dataRT.
|
|
||||||
type DataRTConfig struct {
|
type DataRTConfig struct {
|
||||||
ManualSync ManualSyncConfig `mapstructure:"manual_sync"`
|
Host string `mapstructure:"host"`
|
||||||
|
Port int64 `mapstructure:"port"`
|
||||||
|
PollingAPI string `mapstructure:"polling_api"`
|
||||||
|
Method string `mapstructure:"polling_api_method"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// OtelConfig define config struct of OpenTelemetry tracing
|
// OtelConfig define config struct of OpenTelemetry tracing
|
||||||
|
|
@ -130,9 +124,9 @@ type ModelRTConfig struct {
|
||||||
KafkaConfig `mapstructure:"kafka"`
|
KafkaConfig `mapstructure:"kafka"`
|
||||||
LoggerConfig `mapstructure:"logger"`
|
LoggerConfig `mapstructure:"logger"`
|
||||||
AntsConfig `mapstructure:"ants"`
|
AntsConfig `mapstructure:"ants"`
|
||||||
DataRTConfig DataRTConfig `mapstructure:"dataRT"`
|
DataRTConfig `mapstructure:"dataRT"`
|
||||||
LockerRedisConfig RedisConfig `mapstructure:"locker_redis"`
|
LockerRedisConfig RedisConfig `mapstructure:"locker_redis"`
|
||||||
StorageRedisConfig RedisConfig `mapstructure:"storage_redis"`
|
StorageRedisConfig RedisConfig `mapstructure:"storage_redis"`
|
||||||
AsyncTaskConfig AsyncTaskConfig `mapstructure:"async_task"`
|
AsyncTaskConfig AsyncTaskConfig `mapstructure:"async_task"`
|
||||||
OtelConfig OtelConfig `mapstructure:"otel"`
|
OtelConfig OtelConfig `mapstructure:"otel"`
|
||||||
PostgresDBURI string `mapstructure:"-"`
|
PostgresDBURI string `mapstructure:"-"`
|
||||||
|
|
|
||||||
|
|
@ -1,31 +0,0 @@
|
||||||
package config
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestReadAndInitConfigReadsNestedDataRTManualSync(t *testing.T) {
|
|
||||||
configPath := filepath.Join(t.TempDir(), "config.yaml")
|
|
||||||
contents := []byte(`
|
|
||||||
dataRT:
|
|
||||||
manual_sync:
|
|
||||||
protocol_cl3611_url: "http://127.0.0.1:9001"
|
|
||||||
protocol_104_url: "http://127.0.0.1:9002"
|
|
||||||
api_path: "/api/manual"
|
|
||||||
timeout: 3s
|
|
||||||
`)
|
|
||||||
require.NoError(t, os.WriteFile(configPath, contents, 0o600))
|
|
||||||
|
|
||||||
cfg := ReadAndInitConfig(filepath.Dir(configPath), "config", "yaml")
|
|
||||||
|
|
||||||
assert.Equal(t, "http://127.0.0.1:9001", cfg.DataRTConfig.ManualSync.ProtocolCL3611URL)
|
|
||||||
assert.Equal(t, "http://127.0.0.1:9002", cfg.DataRTConfig.ManualSync.Protocol104URL)
|
|
||||||
assert.Equal(t, "/api/manual", cfg.DataRTConfig.ManualSync.APIPath)
|
|
||||||
assert.Equal(t, 3*time.Second, cfg.DataRTConfig.ManualSync.Timeout)
|
|
||||||
}
|
|
||||||
|
|
@ -17,16 +17,3 @@ const (
|
||||||
// MeasurementModeAutomatic indicates that the measurement runs automatically.
|
// MeasurementModeAutomatic indicates that the measurement runs automatically.
|
||||||
MeasurementModeAutomatic int16 = 1
|
MeasurementModeAutomatic int16 = 1
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
|
||||||
// MeasurementTypeTelemetry represents TM (遥测).
|
|
||||||
MeasurementTypeTelemetry int16 = 0
|
|
||||||
// MeasurementTypeTelesignal represents TS (遥信).
|
|
||||||
MeasurementTypeTelesignal int16 = 1
|
|
||||||
// MeasurementTypeTelecommand represents TC (遥控).
|
|
||||||
MeasurementTypeTelecommand int16 = 2
|
|
||||||
// MeasurementTypeTeleadjusting represents TA (遥调).
|
|
||||||
MeasurementTypeTeleadjusting int16 = 3
|
|
||||||
// MeasurementTypeSetpoint represents SP (定值).
|
|
||||||
MeasurementTypeSetpoint int16 = 4
|
|
||||||
)
|
|
||||||
|
|
|
||||||
|
|
@ -3,9 +3,7 @@ package constants
|
||||||
|
|
||||||
import "strings"
|
import "strings"
|
||||||
|
|
||||||
const ComponentParameterAttributeGroup = "component"
|
var supportedParameterTableSuffixes = [...]string{
|
||||||
|
|
||||||
var supportedDynamicParameterAttributeGroups = [...]string{
|
|
||||||
"base_extend",
|
"base_extend",
|
||||||
"rated",
|
"rated",
|
||||||
"setup",
|
"setup",
|
||||||
|
|
@ -16,32 +14,10 @@ var supportedDynamicParameterAttributeGroups = [...]string{
|
||||||
"behavior",
|
"behavior",
|
||||||
}
|
}
|
||||||
|
|
||||||
// IsSupportedParameterAttributeGroup reports whether token6 identifies a
|
|
||||||
// parameter attribute group supported by the data-object APIs.
|
|
||||||
func IsSupportedParameterAttributeGroup(group string) bool {
|
|
||||||
if group == ComponentParameterAttributeGroup {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
for _, supportedGroup := range supportedDynamicParameterAttributeGroups {
|
|
||||||
if group == supportedGroup {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
// SupportedDynamicParameterAttributeGroups returns the token6 values backed by
|
|
||||||
// project_manager dynamic tables.
|
|
||||||
func SupportedDynamicParameterAttributeGroups() []string {
|
|
||||||
groups := make([]string, len(supportedDynamicParameterAttributeGroups))
|
|
||||||
copy(groups, supportedDynamicParameterAttributeGroups[:])
|
|
||||||
return groups
|
|
||||||
}
|
|
||||||
|
|
||||||
// IsSupportedParameterTableName reports whether a dynamic parameter table has
|
// IsSupportedParameterTableName reports whether a dynamic parameter table has
|
||||||
// one of the supported attribute-group suffixes.
|
// one of the supported attribute-group suffixes.
|
||||||
func IsSupportedParameterTableName(tableName string) bool {
|
func IsSupportedParameterTableName(tableName string) bool {
|
||||||
for _, suffix := range supportedDynamicParameterAttributeGroups {
|
for _, suffix := range supportedParameterTableSuffixes {
|
||||||
if strings.HasSuffix(tableName, "_"+suffix) {
|
if strings.HasSuffix(tableName, "_"+suffix) {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,24 +4,4 @@ package constants
|
||||||
const (
|
const (
|
||||||
// RedisSearchDictName define redis search dictionary name
|
// RedisSearchDictName define redis search dictionary name
|
||||||
RedisSearchDictName = "search_suggestions_dict"
|
RedisSearchDictName = "search_suggestions_dict"
|
||||||
|
|
||||||
// RedisParameterDataObjectKeySet tracks parameter hashes created during
|
|
||||||
// startup so stale parameter data-object keys can be removed safely.
|
|
||||||
RedisParameterDataObjectKeySet = "modelrt:parameter-data-object:keys"
|
|
||||||
|
|
||||||
// RedisMeasurementDataObjectKeySet tracks measurement hashes created during
|
|
||||||
// startup so stale measurement data-object keys can be removed safely.
|
|
||||||
RedisMeasurementDataObjectKeySet = "modelrt:measurement-data-object:keys"
|
|
||||||
|
|
||||||
// RedisParameterDataObjectAliasKeySet tracks parameter alias string keys.
|
|
||||||
RedisParameterDataObjectAliasKeySet = "modelrt:parameter-data-object:alias-keys"
|
|
||||||
|
|
||||||
// RedisParameterDataObjectAliasPrefix prefixes parameter token aliases.
|
|
||||||
RedisParameterDataObjectAliasPrefix = "modelrt:data-object:alias:parameter:"
|
|
||||||
|
|
||||||
// RedisMeasurementDataObjectAliasKeySet tracks measurement alias string keys.
|
|
||||||
RedisMeasurementDataObjectAliasKeySet = "modelrt:measurement-data-object:alias-keys"
|
|
||||||
|
|
||||||
// RedisMeasurementDataObjectAliasPrefix prefixes measurement token aliases.
|
|
||||||
RedisMeasurementDataObjectAliasPrefix = "modelrt:data-object:alias:measurement:"
|
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -1,80 +0,0 @@
|
||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/model"
|
|
||||||
modelsql "modelRT/sql"
|
|
||||||
|
|
||||||
"gorm.io/gorm"
|
|
||||||
)
|
|
||||||
|
|
||||||
// QueryMeasurementInitializationRecords loads every measurement that can be
|
|
||||||
// addressed through the seven-part, four-part, and two-part token forms.
|
|
||||||
func QueryMeasurementInitializationRecords(ctx context.Context, db *gorm.DB) ([]model.MeasurementInitializationRecord, error) {
|
|
||||||
if db == nil {
|
|
||||||
return nil, fmt.Errorf("postgres client is nil")
|
|
||||||
}
|
|
||||||
|
|
||||||
var records []model.MeasurementInitializationRecord
|
|
||||||
if err := db.WithContext(ctx).
|
|
||||||
Raw(compactMeasurementSQL(modelsql.MeasurementInitializationRows)).
|
|
||||||
Scan(&records).Error; err != nil {
|
|
||||||
return nil, fmt.Errorf("query measurement initialization records: %w", err)
|
|
||||||
}
|
|
||||||
if err := validateMeasurementInitializationRecords(records); err != nil {
|
|
||||||
return nil, fmt.Errorf("validate measurement initialization records: %w", err)
|
|
||||||
}
|
|
||||||
return records, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func validateMeasurementInitializationRecords(records []model.MeasurementInitializationRecord) error {
|
|
||||||
for _, record := range records {
|
|
||||||
if record.MeasurementID <= 0 {
|
|
||||||
return fmt.Errorf("measurement %q has invalid id %d", record.MeasurementTag, record.MeasurementID)
|
|
||||||
}
|
|
||||||
if record.ComponentUUID == "" {
|
|
||||||
return fmt.Errorf("measurement %q has empty component uuid", record.MeasurementTag)
|
|
||||||
}
|
|
||||||
if record.GridTag == "" ||
|
|
||||||
record.ZoneTag == "" ||
|
|
||||||
record.StationTag == "" ||
|
|
||||||
record.ComponentNSPath == "" ||
|
|
||||||
record.ComponentTag == "" ||
|
|
||||||
record.MeasurementTag == "" {
|
|
||||||
return fmt.Errorf("measurement %d contains an empty data-object token segment", record.MeasurementID)
|
|
||||||
}
|
|
||||||
if record.MeasurementMode != constants.MeasurementModeManual &&
|
|
||||||
record.MeasurementMode != constants.MeasurementModeAutomatic {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"measurement %q mode must be %d or %d, got %d",
|
|
||||||
record.MeasurementTag,
|
|
||||||
constants.MeasurementModeManual,
|
|
||||||
constants.MeasurementModeAutomatic,
|
|
||||||
record.MeasurementMode,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if _, err := model.MeasurementTypeString(record.MeasurementType); err != nil {
|
|
||||||
return fmt.Errorf("measurement %q: %w", record.MeasurementTag, err)
|
|
||||||
}
|
|
||||||
if record.MeasurementSize <= 0 {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"measurement %q window size must be greater than 0, got %d",
|
|
||||||
record.MeasurementTag,
|
|
||||||
record.MeasurementSize,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if record.MeasurementDataSource == nil {
|
|
||||||
return fmt.Errorf("measurement %q has null data_source", record.MeasurementTag)
|
|
||||||
}
|
|
||||||
if record.MeasurementEventPlan == nil {
|
|
||||||
return fmt.Errorf("measurement %q has null event_plan", record.MeasurementTag)
|
|
||||||
}
|
|
||||||
if record.MeasurementBinding == nil {
|
|
||||||
return fmt.Errorf("measurement %q has null binding", record.MeasurementTag)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,138 +0,0 @@
|
||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"modelRT/model"
|
|
||||||
modelsql "modelRT/sql"
|
|
||||||
|
|
||||||
"github.com/DATA-DOG/go-sqlmock"
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
"gorm.io/driver/postgres"
|
|
||||||
"gorm.io/gorm"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestMeasurementInitializationSQLJoinsRequiredHierarchy(t *testing.T) {
|
|
||||||
statement := compactMeasurementSQL(modelsql.MeasurementInitializationRows)
|
|
||||||
|
|
||||||
assert.Contains(t, statement, "component.station_id = station.id")
|
|
||||||
assert.Contains(t, statement, "measurement.component_uuid = component.global_uuid")
|
|
||||||
assert.Contains(t, statement, "bay.bay_uuid = measurement.bay_uuid")
|
|
||||||
assert.Contains(t, statement, "measurement.type AS measurement_type")
|
|
||||||
assert.Contains(t, statement, "measurement.tag <> ''")
|
|
||||||
assert.NotContains(t, strings.ToLower(statement), "dev_")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestQueryMeasurementInitializationRecords(t *testing.T) {
|
|
||||||
sqlDB, mock, err := sqlmock.New()
|
|
||||||
require.NoError(t, err)
|
|
||||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
|
||||||
|
|
||||||
db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{})
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
mock.ExpectQuery(`(?s)SELECT.*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*INNER JOIN public\.measurement.*INNER JOIN public\.bay`).
|
|
||||||
WillReturnRows(measurementInitializationRows().
|
|
||||||
AddRow(
|
|
||||||
"grid000",
|
|
||||||
"zone000",
|
|
||||||
"station000",
|
|
||||||
"component-uuid",
|
|
||||||
"nspath",
|
|
||||||
"component",
|
|
||||||
int64(10),
|
|
||||||
"IA_rms",
|
|
||||||
"A相保护电流有效值",
|
|
||||||
int16(0),
|
|
||||||
int16(1),
|
|
||||||
1,
|
|
||||||
`{"type":1,"io_address":{"channel":"TM1"}}`,
|
|
||||||
`{}`,
|
|
||||||
`{"ct":{"ratio":1250}}`,
|
|
||||||
))
|
|
||||||
|
|
||||||
records, err := QueryMeasurementInitializationRecords(context.Background(), db)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Len(t, records, 1)
|
|
||||||
assert.Equal(t, int64(10), records[0].MeasurementID)
|
|
||||||
assert.Equal(t, "IA_rms", records[0].MeasurementTag)
|
|
||||||
assert.Equal(t, int16(0), records[0].MeasurementType)
|
|
||||||
assert.Equal(t, float64(1), records[0].MeasurementDataSource["type"])
|
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestValidateMeasurementInitializationRecords(t *testing.T) {
|
|
||||||
record := validMeasurementInitializationRecord()
|
|
||||||
|
|
||||||
invalidMode := record
|
|
||||||
invalidMode.MeasurementMode = 3
|
|
||||||
err := validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{invalidMode})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "mode must be 0 or 1")
|
|
||||||
|
|
||||||
emptySegment := record
|
|
||||||
emptySegment.ComponentNSPath = ""
|
|
||||||
err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{emptySegment})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "empty data-object token segment")
|
|
||||||
|
|
||||||
nullDataSource := record
|
|
||||||
nullDataSource.MeasurementDataSource = nil
|
|
||||||
err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{nullDataSource})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "null data_source")
|
|
||||||
|
|
||||||
invalidType := record
|
|
||||||
invalidType.MeasurementType = -1
|
|
||||||
err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{invalidType})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "unsupported measurement type -1")
|
|
||||||
|
|
||||||
invalidSize := record
|
|
||||||
invalidSize.MeasurementSize = 0
|
|
||||||
err = validateMeasurementInitializationRecords([]model.MeasurementInitializationRecord{invalidSize})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "window size must be greater than 0")
|
|
||||||
}
|
|
||||||
|
|
||||||
func measurementInitializationRows() *sqlmock.Rows {
|
|
||||||
return sqlmock.NewRows([]string{
|
|
||||||
"grid_tag",
|
|
||||||
"zone_tag",
|
|
||||||
"station_tag",
|
|
||||||
"component_uuid",
|
|
||||||
"component_nspath",
|
|
||||||
"component_tag",
|
|
||||||
"measurement_id",
|
|
||||||
"measurement_tag",
|
|
||||||
"measurement_name",
|
|
||||||
"measurement_type",
|
|
||||||
"measurement_mode",
|
|
||||||
"measurement_size",
|
|
||||||
"measurement_data_source",
|
|
||||||
"measurement_event_plan",
|
|
||||||
"measurement_binding",
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
func validMeasurementInitializationRecord() model.MeasurementInitializationRecord {
|
|
||||||
return model.MeasurementInitializationRecord{
|
|
||||||
GridTag: "grid",
|
|
||||||
ZoneTag: "zone",
|
|
||||||
StationTag: "station",
|
|
||||||
ComponentUUID: "component-uuid",
|
|
||||||
ComponentNSPath: "nspath",
|
|
||||||
ComponentTag: "component",
|
|
||||||
MeasurementID: 1,
|
|
||||||
MeasurementTag: "measurement",
|
|
||||||
MeasurementType: 0,
|
|
||||||
MeasurementMode: 1,
|
|
||||||
MeasurementSize: 1,
|
|
||||||
MeasurementDataSource: map[string]any{},
|
|
||||||
MeasurementEventPlan: map[string]any{},
|
|
||||||
MeasurementBinding: map[string]any{},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,141 +0,0 @@
|
||||||
// Package database define database operation functions
|
|
||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/model"
|
|
||||||
"modelRT/sql"
|
|
||||||
|
|
||||||
"gorm.io/gorm"
|
|
||||||
)
|
|
||||||
|
|
||||||
type parameterInitializationRoute struct {
|
|
||||||
TableName string `gorm:"column:name"`
|
|
||||||
ModelName string `gorm:"column:tag"`
|
|
||||||
AttributeGroup string `gorm:"column:group_name"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// QueryParameterInitializationRecords loads every parameter accepted by the
|
|
||||||
// data-object query API. Dynamic parameters are resolved through
|
|
||||||
// project_manager; component parameters are read directly from component.
|
|
||||||
func QueryParameterInitializationRecords(ctx context.Context, db *gorm.DB) ([]model.ParameterInitializationRecord, error) {
|
|
||||||
if db == nil {
|
|
||||||
return nil, fmt.Errorf("postgres client is nil")
|
|
||||||
}
|
|
||||||
|
|
||||||
var routes []parameterInitializationRoute
|
|
||||||
if err := db.WithContext(ctx).
|
|
||||||
Raw(
|
|
||||||
compactParameterSQL(sql.ParameterInitializationRoutes),
|
|
||||||
constants.SupportedDynamicParameterAttributeGroups(),
|
|
||||||
).
|
|
||||||
Scan(&routes).Error; err != nil {
|
|
||||||
return nil, fmt.Errorf("query parameter initialization routes: %w", err)
|
|
||||||
}
|
|
||||||
if err := validateParameterInitializationRoutes(routes); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
records := make([]model.ParameterInitializationRecord, 0)
|
|
||||||
for _, route := range routes {
|
|
||||||
quotedTableName := `"` + route.TableName + `"`
|
|
||||||
query := compactParameterSQL(
|
|
||||||
fmt.Sprintf(sql.DynamicParameterInitializationRows, quotedTableName),
|
|
||||||
)
|
|
||||||
|
|
||||||
var tableRecords []model.ParameterInitializationRecord
|
|
||||||
if err := db.WithContext(ctx).
|
|
||||||
Raw(query, route.TableName, route.ModelName, route.AttributeGroup).
|
|
||||||
Scan(&tableRecords).Error; err != nil {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"query parameter initialization table %q for model %q group %q: %w",
|
|
||||||
route.TableName,
|
|
||||||
route.ModelName,
|
|
||||||
route.AttributeGroup,
|
|
||||||
err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if err := validateParameterInitializationRecords(tableRecords); err != nil {
|
|
||||||
return nil, fmt.Errorf("validate parameter initialization table %q: %w", route.TableName, err)
|
|
||||||
}
|
|
||||||
records = append(records, tableRecords...)
|
|
||||||
}
|
|
||||||
|
|
||||||
var componentRecords []model.ParameterInitializationRecord
|
|
||||||
if err := db.WithContext(ctx).
|
|
||||||
Raw(compactParameterSQL(sql.ComponentParameterInitializationRows)).
|
|
||||||
Scan(&componentRecords).Error; err != nil {
|
|
||||||
return nil, fmt.Errorf("query component parameter initialization records: %w", err)
|
|
||||||
}
|
|
||||||
if err := validateParameterInitializationRecords(componentRecords); err != nil {
|
|
||||||
return nil, fmt.Errorf("validate component parameter initialization records: %w", err)
|
|
||||||
}
|
|
||||||
records = append(records, componentRecords...)
|
|
||||||
return records, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func validateParameterInitializationRoutes(routes []parameterInitializationRoute) error {
|
|
||||||
seen := make(map[string]struct{}, len(routes))
|
|
||||||
for _, route := range routes {
|
|
||||||
if !validParameterTableName(route.TableName) {
|
|
||||||
return fmt.Errorf("project_manager contains unsupported parameter table name %q", route.TableName)
|
|
||||||
}
|
|
||||||
if !constants.IsSupportedParameterAttributeGroup(route.AttributeGroup) ||
|
|
||||||
route.AttributeGroup == constants.ComponentParameterAttributeGroup {
|
|
||||||
return fmt.Errorf("project_manager contains unsupported dynamic attribute group %q", route.AttributeGroup)
|
|
||||||
}
|
|
||||||
|
|
||||||
key := route.ModelName + "\x00" + route.AttributeGroup
|
|
||||||
if _, exists := seen[key]; exists {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"model %q and attribute group %q match more than one project_manager record",
|
|
||||||
route.ModelName,
|
|
||||||
route.AttributeGroup,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
seen[key] = struct{}{}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func validateParameterInitializationRecords(records []model.ParameterInitializationRecord) error {
|
|
||||||
for _, record := range records {
|
|
||||||
switch record.DynamicRecordCount {
|
|
||||||
case 1:
|
|
||||||
case 0:
|
|
||||||
return fmt.Errorf(
|
|
||||||
"component %q has no %q parameter record",
|
|
||||||
record.ComponentTag,
|
|
||||||
record.AttributeGroup,
|
|
||||||
)
|
|
||||||
default:
|
|
||||||
return fmt.Errorf(
|
|
||||||
"component %q has %d %q parameter records",
|
|
||||||
record.ComponentTag,
|
|
||||||
record.DynamicRecordCount,
|
|
||||||
record.AttributeGroup,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if record.AttributeName == "" || record.AttributeType == "" {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"component %q group %q contains an invalid parameter column",
|
|
||||||
record.ComponentTag,
|
|
||||||
record.AttributeGroup,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
switch record.DescriptionCount {
|
|
||||||
case 0:
|
|
||||||
return fmt.Errorf("parameter description not found for attribute %q", record.AttributeName)
|
|
||||||
case 1:
|
|
||||||
if !record.Description.Valid {
|
|
||||||
return fmt.Errorf("parameter description is null for attribute %q", record.AttributeName)
|
|
||||||
}
|
|
||||||
default:
|
|
||||||
return fmt.Errorf("ambiguous parameter description for attribute %q", record.AttributeName)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,158 +0,0 @@
|
||||||
// Package database define database operation functions
|
|
||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"database/sql"
|
|
||||||
"database/sql/driver"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/model"
|
|
||||||
modelsql "modelRT/sql"
|
|
||||||
|
|
||||||
"github.com/DATA-DOG/go-sqlmock"
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
"gorm.io/driver/postgres"
|
|
||||||
"gorm.io/gorm"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestParameterInitializationSQLUsesStationIDAndExcludesItFromComponentAttributes(t *testing.T) {
|
|
||||||
dynamicSQL := compactParameterSQL(modelsql.DynamicParameterInitializationRows)
|
|
||||||
componentSQL := compactParameterSQL(modelsql.ComponentParameterInitializationRows)
|
|
||||||
|
|
||||||
assert.Contains(t, dynamicSQL, "component.station_id = station.id")
|
|
||||||
assert.Contains(t, componentSQL, "component.station_id = station.id")
|
|
||||||
assert.Contains(t, componentSQL, "to_jsonb(component) - 'station_id'")
|
|
||||||
assert.Contains(t, dynamicSQL, "component.nspath <> ''")
|
|
||||||
assert.Contains(t, dynamicSQL, "component.tag <> ''")
|
|
||||||
assert.Contains(t, componentSQL, "component.nspath <> ''")
|
|
||||||
assert.Contains(t, componentSQL, "component.tag <> ''")
|
|
||||||
assert.NotContains(t, strings.ToLower(componentSQL), "component.station = station.tagname")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestQueryParameterInitializationRecordsJoinsHierarchyAndDynamicTable(t *testing.T) {
|
|
||||||
sqlDB, mock, err := sqlmock.New()
|
|
||||||
require.NoError(t, err)
|
|
||||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
|
||||||
|
|
||||||
db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{})
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
groups := constants.SupportedDynamicParameterAttributeGroups()
|
|
||||||
routeArgs := make([]driver.Value, len(groups))
|
|
||||||
for index, group := range groups {
|
|
||||||
routeArgs[index] = group
|
|
||||||
}
|
|
||||||
mock.ExpectQuery(`(?s)SELECT name, tag, group_name.*FROM project_manager.*WHERE group_name IN`).
|
|
||||||
WithArgs(routeArgs...).
|
|
||||||
WillReturnRows(sqlmock.NewRows([]string{"name", "tag", "group_name"}).
|
|
||||||
AddRow("cable_cable_demo_base_extend", "cable_demo", "base_extend"))
|
|
||||||
|
|
||||||
mock.ExpectQuery(`(?s)WITH dynamic_rows AS.*FROM public\."cable_cable_demo_base_extend".*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*INNER JOIN public\.project_manager.*jsonb_each`).
|
|
||||||
WithArgs("cable_cable_demo_base_extend", "cable_demo", "base_extend").
|
|
||||||
WillReturnRows(parameterInitializationRows().
|
|
||||||
AddRow(
|
|
||||||
"grid000",
|
|
||||||
"zone000",
|
|
||||||
"station000",
|
|
||||||
true,
|
|
||||||
"component-uuid",
|
|
||||||
"nspath",
|
|
||||||
"component",
|
|
||||||
"base_extend",
|
|
||||||
"vnom_kv",
|
|
||||||
"220.0",
|
|
||||||
"DOUBLE PRECISION",
|
|
||||||
"额定电压",
|
|
||||||
int64(1),
|
|
||||||
int64(1),
|
|
||||||
))
|
|
||||||
|
|
||||||
mock.ExpectQuery(`(?s)SELECT.*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*jsonb_each`).
|
|
||||||
WillReturnRows(parameterInitializationRows().
|
|
||||||
AddRow(
|
|
||||||
"grid000",
|
|
||||||
"zone000",
|
|
||||||
"station000",
|
|
||||||
true,
|
|
||||||
"component-uuid",
|
|
||||||
"nspath",
|
|
||||||
"component",
|
|
||||||
"component",
|
|
||||||
"description",
|
|
||||||
`"组件"`,
|
|
||||||
"CHARACTER VARYING(512)",
|
|
||||||
"组件名称",
|
|
||||||
int64(1),
|
|
||||||
int64(1),
|
|
||||||
))
|
|
||||||
|
|
||||||
records, err := QueryParameterInitializationRecords(context.Background(), db)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Len(t, records, 2)
|
|
||||||
assert.Equal(t, "vnom_kv", records[0].AttributeName)
|
|
||||||
assert.Equal(t, "description", records[1].AttributeName)
|
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestValidateParameterInitializationRoutesRejectsAmbiguousMapping(t *testing.T) {
|
|
||||||
routes := []parameterInitializationRoute{
|
|
||||||
{TableName: "cable_demo_stable", ModelName: "cable_demo", AttributeGroup: "stable"},
|
|
||||||
{TableName: "cable_other_stable", ModelName: "cable_demo", AttributeGroup: "stable"},
|
|
||||||
}
|
|
||||||
|
|
||||||
err := validateParameterInitializationRoutes(routes)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "more than one project_manager record")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestValidateParameterInitializationRecordsEnforcesDescriptionAndRowUniqueness(t *testing.T) {
|
|
||||||
validRecord := modelParameterInitializationRecordForTest()
|
|
||||||
|
|
||||||
missingDescription := validRecord
|
|
||||||
missingDescription.Description = sql.NullString{}
|
|
||||||
missingDescription.DescriptionCount = 0
|
|
||||||
err := validateParameterInitializationRecords([]model.ParameterInitializationRecord{missingDescription})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "description not found")
|
|
||||||
|
|
||||||
duplicateRow := validRecord
|
|
||||||
duplicateRow.DynamicRecordCount = 2
|
|
||||||
err = validateParameterInitializationRecords([]model.ParameterInitializationRecord{duplicateRow})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "has 2")
|
|
||||||
}
|
|
||||||
|
|
||||||
func parameterInitializationRows() *sqlmock.Rows {
|
|
||||||
return sqlmock.NewRows([]string{
|
|
||||||
"grid_tag",
|
|
||||||
"zone_tag",
|
|
||||||
"station_tag",
|
|
||||||
"station_is_local",
|
|
||||||
"component_uuid",
|
|
||||||
"component_nspath",
|
|
||||||
"component_tag",
|
|
||||||
"attribute_group",
|
|
||||||
"attribute_name",
|
|
||||||
"attribute_value",
|
|
||||||
"attribute_type",
|
|
||||||
"description",
|
|
||||||
"description_count",
|
|
||||||
"dynamic_record_count",
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
func modelParameterInitializationRecordForTest() model.ParameterInitializationRecord {
|
|
||||||
return model.ParameterInitializationRecord{
|
|
||||||
ComponentTag: "component",
|
|
||||||
AttributeGroup: "stable",
|
|
||||||
AttributeName: "attribute",
|
|
||||||
AttributeType: "INTEGER",
|
|
||||||
Description: sql.NullString{String: "属性", Valid: true},
|
|
||||||
DescriptionCount: 1,
|
|
||||||
DynamicRecordCount: 1,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -439,10 +439,10 @@ go run deploy/redis-test-data/measurments-recommend/measurement_injection.go
|
||||||
| | `station_id` | 项目所操作的默认变电站 `ID`。 | `1` |
|
| | `station_id` | 项目所操作的默认变电站 `ID`。 | `1` |
|
||||||
| **Service Config** | `service_name` | 服务名称，用于日志、监控等标识。 | `"modelRT"` |
|
| **Service Config** | `service_name` | 服务名称，用于日志、监控等标识。 | `"modelRT"` |
|
||||||
| | `secret_key` | 服务内部使用的秘钥，用于签名或认证。 | `"modelrt_key"` |
|
| | `secret_key` | 服务内部使用的秘钥，用于签名或认证。 | `"modelrt_key"` |
|
||||||
| **DataRT Manual Sync** | `manual_sync.protocol_cl3611_url` | CL3611 协议服务地址。 | `"http://127.0.0.1:9001"` |
|
| **DataRT API** | `host` | 外部 `DataRT` 服务的主机地址。 | `"http://127.0.0.1"` |
|
||||||
| | `manual_sync.protocol_104_url` | IEC 60870-5-104 协议服务地址。 | `"http://127.0.0.1:9002"` |
|
| | `port` | `DataRT` 服务的端口号。 | `8888` |
|
||||||
| | `manual_sync.api_path` | 手动测量值及模式同步 API 路径。 | `"/api/manual"` |
|
| | `polling_api` | 轮询数据的 `API` 路径。 | `"datart/getPointData"` |
|
||||||
| | `manual_sync.timeout` | 同步请求超时时间。 | `"3s"` |
|
| | `polling_api_method` | 调用该 `API` 使用的 `HTTP` 方法。 | `"GET"` |
|
||||||
|
|
||||||
#### 3.2 编译 ModelRT 服务
|
#### 3.2 编译 ModelRT 服务
|
||||||
|
|
||||||
|
|
@ -759,94 +759,6 @@ kubectl delete -f deploy/k8s/pg-service.yaml \
|
||||||
-f deploy/k8s/pg-configmap.yaml
|
-f deploy/k8s/pg-configmap.yaml
|
||||||
```
|
```
|
||||||
|
|
||||||
#### 4.5 部署 MongoDB 并创建应用用户
|
|
||||||
|
|
||||||
使用以下清单部署 MongoDB：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl apply -f deploy/k8s/mongodb-secret.yaml
|
|
||||||
kubectl apply -f deploy/k8s/mongodb-pvc.yaml
|
|
||||||
kubectl apply -f deploy/k8s/mongodb-statefulset.yaml
|
|
||||||
kubectl apply -f deploy/k8s/mongodb-service.yaml
|
|
||||||
```
|
|
||||||
|
|
||||||
等待 MongoDB Pod 就绪：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl wait --for=condition=ready pod/mongodb-0 --timeout=180s
|
|
||||||
```
|
|
||||||
|
|
||||||
MongoDB 首次初始化时会根据 `mongodb-secret.yaml` 创建 `admin` 管理员。Pod 就绪后，以管理员身份在 `admin` 认证库中创建应用用户 `coslight`，并授予其对 `eventdb` 的读写和数据库管理权限：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl exec mongodb-0 -- mongosh \
|
|
||||||
-u admin \
|
|
||||||
-p coslight \
|
|
||||||
--authenticationDatabase admin \
|
|
||||||
--quiet \
|
|
||||||
--eval '
|
|
||||||
const adminDb = db.getSiblingDB("admin");
|
|
||||||
adminDb.createUser({
|
|
||||||
user: "coslight",
|
|
||||||
pwd: "coslight",
|
|
||||||
roles: [
|
|
||||||
{ role: "readWrite", db: "eventdb" },
|
|
||||||
{ role: "dbAdmin", db: "eventdb" }
|
|
||||||
]
|
|
||||||
});
|
|
||||||
'
|
|
||||||
```
|
|
||||||
|
|
||||||
> **注意：** `use admin` 是 `mongosh` 的交互式命令，不应在 `--eval` 脚本中使用。这里通过 `db.getSiblingDB("admin")` 明确指定用户所属的认证库。用户存储在 `admin` 库中，因此应用连接时必须将认证库配置为 `admin`。
|
|
||||||
|
|
||||||
检查用户是否创建成功：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl exec mongodb-0 -- mongosh \
|
|
||||||
-u admin \
|
|
||||||
-p coslight \
|
|
||||||
--authenticationDatabase admin \
|
|
||||||
--quiet \
|
|
||||||
--eval 'printjson(db.getSiblingDB("admin").getUser("coslight"));'
|
|
||||||
```
|
|
||||||
|
|
||||||
使用 `coslight` 用户连接并验证 `eventdb` 权限：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl exec mongodb-0 -- mongosh \
|
|
||||||
-u coslight \
|
|
||||||
-p coslight \
|
|
||||||
--authenticationDatabase admin \
|
|
||||||
--quiet \
|
|
||||||
--eval '
|
|
||||||
const eventDb = db.getSiblingDB("eventdb");
|
|
||||||
eventDb.__permission_check.insertOne({ checkedAt: new Date() });
|
|
||||||
eventDb.__permission_check.deleteMany({});
|
|
||||||
printjson({ ok: 1, database: eventDb.getName() });
|
|
||||||
'
|
|
||||||
```
|
|
||||||
|
|
||||||
如果 `coslight` 用户已经存在，重复执行 `createUser` 会返回 `User already exists`。需要重置密码或修正角色时，使用：
|
|
||||||
|
|
||||||
```bash
|
|
||||||
kubectl exec mongodb-0 -- mongosh \
|
|
||||||
-u admin \
|
|
||||||
-p coslight \
|
|
||||||
--authenticationDatabase admin \
|
|
||||||
--quiet \
|
|
||||||
--eval '
|
|
||||||
db.getSiblingDB("admin").updateUser("coslight", {
|
|
||||||
pwd: "coslight",
|
|
||||||
roles: [
|
|
||||||
{ role: "readWrite", db: "eventdb" },
|
|
||||||
{ role: "dbAdmin", db: "eventdb" }
|
|
||||||
]
|
|
||||||
});
|
|
||||||
'
|
|
||||||
```
|
|
||||||
|
|
||||||
> **安全提示：** 示例使用仓库当前的测试密码。生产环境应修改管理员和应用用户密码，并避免在命令行或版本库中保存明文凭据。
|
|
||||||
|
|
||||||
### 5\. 部署 ModelRT（Kubernetes）
|
### 5\. 部署 ModelRT（Kubernetes）
|
||||||
|
|
||||||
所有资源部署在 `default` 命名空间，YAML 文件位于 `deploy/k8s/`。
|
所有资源部署在 `default` 命名空间，YAML 文件位于 `deploy/k8s/`。
|
||||||
|
|
|
||||||
|
|
@ -80,9 +80,7 @@ data:
|
||||||
deploy_env: "development"
|
deploy_env: "development"
|
||||||
|
|
||||||
dataRT:
|
dataRT:
|
||||||
# manual measurement synchronization endpoints
|
host: "http://127.0.0.1"
|
||||||
manual_sync:
|
port: 8888
|
||||||
protocol_cl3611_url: "http://protocol-cl3611-service:9001"
|
polling_api: "datart/getPointData"
|
||||||
protocol_104_url: "http://protocol-104-service:9002"
|
polling_api_method: "GET"
|
||||||
api_path: "/api/manual"
|
|
||||||
timeout: 3s
|
|
||||||
|
|
|
||||||
|
|
@ -61,9 +61,8 @@ func ProcessMeasurements(measurements []orm.Measurement) map[string]CalculationR
|
||||||
station, _ := ioAddress["station"].(string)
|
station, _ := ioAddress["station"].(string)
|
||||||
device, _ := ioAddress["device"].(string)
|
device, _ := ioAddress["device"].(string)
|
||||||
channel, _ := ioAddress["channel"].(string)
|
channel, _ := ioAddress["channel"].(string)
|
||||||
option, _ := ioAddress["option"].(string)
|
|
||||||
|
|
||||||
result := strings.ToLower(fmt.Sprintf("%s:%s:phasor:%s:%s", station, device, channel, option))
|
result := strings.ToLower(fmt.Sprintf("%s:%s:phasor:%s", station, device, channel))
|
||||||
if measurement.EventPlan == nil {
|
if measurement.EventPlan == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,7 +4,6 @@ package diagram
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
"github.com/redis/go-redis/v9"
|
||||||
|
|
@ -19,73 +18,40 @@ type RedisClient struct {
|
||||||
// greatest numeric timestamp. Measurement ZSets currently store timestamp in
|
// greatest numeric timestamp. Measurement ZSets currently store timestamp in
|
||||||
// member and measurement value in score.
|
// member and measurement value in score.
|
||||||
func (rc *RedisClient) QueryLatestMeasurementValue(ctx context.Context, key string) (float64, error) {
|
func (rc *RedisClient) QueryLatestMeasurementValue(ctx context.Context, key string) (float64, error) {
|
||||||
values, err := rc.QueryLatestMeasurementValues(ctx, key, 1)
|
|
||||||
if err != nil {
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
return values[0], nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// QueryLatestMeasurementValues returns up to size scores ordered by their
|
|
||||||
// numeric member timestamps from newest to oldest.
|
|
||||||
func (rc *RedisClient) QueryLatestMeasurementValues(ctx context.Context, key string, size int) ([]float64, error) {
|
|
||||||
if rc.Client == nil {
|
if rc.Client == nil {
|
||||||
return nil, fmt.Errorf("redis client is not initialized")
|
return 0, fmt.Errorf("redis client is not initialized")
|
||||||
}
|
|
||||||
if size <= 0 {
|
|
||||||
return nil, fmt.Errorf("measurement window size must be greater than 0, got %d", size)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
members, err := rc.Client.ZRangeWithScores(ctx, key, 0, -1).Result()
|
members, err := rc.Client.ZRangeWithScores(ctx, key, 0, -1).Result()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return 0, err
|
||||||
}
|
}
|
||||||
return latestMeasurementValues(members, key, size)
|
return latestMeasurementValue(members, key)
|
||||||
}
|
}
|
||||||
|
|
||||||
func latestMeasurementValue(members []redis.Z, key string) (float64, error) {
|
func latestMeasurementValue(members []redis.Z, key string) (float64, error) {
|
||||||
values, err := latestMeasurementValues(members, key, 1)
|
|
||||||
if err != nil {
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
return values[0], nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func latestMeasurementValues(members []redis.Z, key string, size int) ([]float64, error) {
|
|
||||||
if size <= 0 {
|
|
||||||
return nil, fmt.Errorf("measurement window size must be greater than 0, got %d", size)
|
|
||||||
}
|
|
||||||
if len(members) == 0 {
|
if len(members) == 0 {
|
||||||
return nil, fmt.Errorf("real-time measurement value not found for key %q", key)
|
return 0, fmt.Errorf("real-time measurement value not found for key %q", key)
|
||||||
}
|
}
|
||||||
|
|
||||||
type timestampedValue struct {
|
var latestTimestamp int64
|
||||||
timestamp int64
|
var latestValue float64
|
||||||
value float64
|
found := false
|
||||||
}
|
|
||||||
values := make([]timestampedValue, 0, len(members))
|
|
||||||
for _, member := range members {
|
for _, member := range members {
|
||||||
timestamp, err := strconv.ParseInt(fmt.Sprint(member.Member), 10, 64)
|
timestamp, err := strconv.ParseInt(fmt.Sprint(member.Member), 10, 64)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
values = append(values, timestampedValue{timestamp: timestamp, value: member.Score})
|
if !found || timestamp > latestTimestamp {
|
||||||
|
latestTimestamp = timestamp
|
||||||
|
latestValue = member.Score
|
||||||
|
found = true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if len(values) == 0 {
|
if !found {
|
||||||
return nil, fmt.Errorf("real-time measurement timestamps are invalid for key %q", key)
|
return 0, fmt.Errorf("real-time measurement timestamps are invalid for key %q", key)
|
||||||
}
|
}
|
||||||
|
return latestValue, nil
|
||||||
sort.Slice(values, func(i, j int) bool {
|
|
||||||
return values[i].timestamp > values[j].timestamp
|
|
||||||
})
|
|
||||||
if size > len(values) {
|
|
||||||
size = len(values)
|
|
||||||
}
|
|
||||||
result := make([]float64, size)
|
|
||||||
for index := range size {
|
|
||||||
result[index] = values[index].value
|
|
||||||
}
|
|
||||||
return result, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewRedisClient define func of new redis client instance
|
// NewRedisClient define func of new redis client instance
|
||||||
|
|
|
||||||
|
|
@ -26,32 +26,3 @@ func TestLatestMeasurementValueRejectsMissingOrInvalidTimestamps(t *testing.T) {
|
||||||
_, err = latestMeasurementValue([]redis.Z{{Member: "invalid", Score: 1}}, "measurement-key")
|
_, err = latestMeasurementValue([]redis.Z{{Member: "invalid", Score: 1}}, "measurement-key")
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestLatestMeasurementValuesReturnsNewestWindow(t *testing.T) {
|
|
||||||
values, err := latestMeasurementValues([]redis.Z{
|
|
||||||
{Member: "100", Score: 10},
|
|
||||||
{Member: "400", Score: 40},
|
|
||||||
{Member: "invalid", Score: 999},
|
|
||||||
{Member: "200", Score: 20},
|
|
||||||
{Member: "300", Score: 30},
|
|
||||||
}, "measurement-key", 3)
|
|
||||||
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, []float64{40, 30, 20}, values)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestLatestMeasurementValuesReturnsAvailableWindow(t *testing.T) {
|
|
||||||
values, err := latestMeasurementValues([]redis.Z{
|
|
||||||
{Member: "100", Score: 10},
|
|
||||||
{Member: "200", Score: 20},
|
|
||||||
}, "measurement-key", 5)
|
|
||||||
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, []float64{20, 10}, values)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestLatestMeasurementValuesRejectsInvalidSize(t *testing.T) {
|
|
||||||
_, err := latestMeasurementValues([]redis.Z{{Member: "100", Score: 10}}, "measurement-key", 0)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "window size must be greater than 0")
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -3,27 +3,26 @@ package handler
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strconv"
|
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"modelRT/common"
|
"modelRT/common"
|
||||||
"modelRT/common/errcode"
|
"modelRT/common/errcode"
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
|
"modelRT/database"
|
||||||
"modelRT/diagram"
|
"modelRT/diagram"
|
||||||
"modelRT/logger"
|
"modelRT/logger"
|
||||||
"modelRT/model"
|
"modelRT/model"
|
||||||
"modelRT/orm"
|
"modelRT/orm"
|
||||||
|
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// DataObjectAttributeQueryHandler define data object attribute value query process API
|
// DataObjectAttributeQueryHandler define data object attribute value query process API
|
||||||
func DataObjectAttributeQueryHandler(c *gin.Context) {
|
func DataObjectAttributeQueryHandler(c *gin.Context) {
|
||||||
ctx := c.Request.Context()
|
ctx := c.Request.Context()
|
||||||
|
pgClient := database.GetPostgresDBClient()
|
||||||
|
|
||||||
token, field, err := parseDataObjectAttributeQuery(c)
|
token, field, err := parseDataObjectAttributeQuery(c)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -45,37 +44,112 @@ func DataObjectAttributeQueryHandler(c *gin.Context) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
value, err := queryDataObjectAttributeValue(
|
var parameter *database.ParameterDataObject
|
||||||
ctx,
|
var measurement *orm.Measurement
|
||||||
dataObjectType,
|
var measurementComponent *orm.Component
|
||||||
token,
|
switch dataObjectType {
|
||||||
field,
|
case constants.DataObjectTypeParameter:
|
||||||
loadDataObjectHashField,
|
// 参量支持两种形式token4.token5.token6.token7与token1.token2.token3.token4.token5.token6.token7
|
||||||
loadMeasurementValueMetadata,
|
parameter, err = database.QueryParameterByDataObjectToken(ctx, pgClient, token)
|
||||||
queryMeasurementRealtimeValue,
|
if err != nil {
|
||||||
)
|
if errors.Is(err, common.ErrInvalidParameterToken) ||
|
||||||
if err != nil {
|
errors.Is(err, common.ErrParameterTokenNotFound) ||
|
||||||
if isDataObjectTokenNotFound(err) {
|
errors.Is(err, common.ErrAmbiguousParameterToken) {
|
||||||
logger.Warn(ctx, "query data-object token from redis failed", "token", token, "field", field, "error", err)
|
logger.Warn(ctx, "validate parameter token failed", "token", token, "error", err)
|
||||||
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Error(ctx, "query parameter token from postgres failed", "token", token, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeServerError, "validate parameter token failed", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
case constants.DataObjectTypeMeasurement:
|
||||||
|
// 量测支持token1.token2.token3.token4.token5.token6.token7、token4.token5.token6.token7、token4.token7
|
||||||
|
measurement, measurementComponent, err = database.QueryMeasurementByDataObjectToken(ctx, pgClient, token)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, common.ErrInvalidMeasurementToken) ||
|
||||||
|
errors.Is(err, common.ErrMeasurementTokenNotFound) ||
|
||||||
|
errors.Is(err, common.ErrAmbiguousMeasurementToken) {
|
||||||
|
logger.Warn(ctx, "validate measurement token failed", "token", token, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Error(ctx, "query measurement token from postgres failed", "token", token, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeServerError, "validate measurement token failed", nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
switch dataObjectType {
|
||||||
|
case constants.DataObjectTypeParameter:
|
||||||
|
value, err := buildParameterAttributeValue(
|
||||||
|
ctx,
|
||||||
|
field,
|
||||||
|
parameter,
|
||||||
|
func(ctx context.Context, parameter *database.ParameterDataObject) (any, error) {
|
||||||
|
return database.QueryParameterDataObjectValue(ctx, pgClient, parameter)
|
||||||
|
},
|
||||||
|
func(ctx context.Context, attributeName string) (string, error) {
|
||||||
|
return database.QueryParameterAttributeDescription(ctx, pgClient, attributeName)
|
||||||
|
},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, common.ErrUnsupportedParameterField) {
|
||||||
|
logger.Warn(ctx, "query unsupported parameter field", "token", token, "field", field, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.Error(ctx, "build parameter attribute value failed", "token", token, "field", field, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeServerError, "query parameter attribute failed", nil)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Error(ctx, "query data-object attribute from redis failed", "token", token, "field", field, "error", err)
|
result := dataObjectAttributeQueryResult{
|
||||||
renderRespFailure(c, constants.RespCodeServerError, dataObjectAttributeFailureMessage(dataObjectType), nil)
|
Token: token,
|
||||||
return
|
Field: field,
|
||||||
}
|
Code: errcode.ErrProcessSuccess.Code(),
|
||||||
|
Msg: errcode.ErrProcessSuccess.Msg(),
|
||||||
|
Value: value,
|
||||||
|
}
|
||||||
|
renderRespSuccess(c, constants.RespCodeSuccess, "query parameter attribute success", map[string]any{
|
||||||
|
"attributes": []dataObjectAttributeQueryResult{result},
|
||||||
|
})
|
||||||
|
case constants.DataObjectTypeMeasurement:
|
||||||
|
value, err := buildMeasurementAttributeValue(
|
||||||
|
ctx,
|
||||||
|
field,
|
||||||
|
measurement,
|
||||||
|
measurementComponent,
|
||||||
|
queryMeasurementRealtimeValue,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
if errors.Is(err, common.ErrUnsupportedMeasurementField) {
|
||||||
|
logger.Warn(ctx, "query unsupported measurement field", "token", token, "field", field, "error", err)
|
||||||
|
renderRespFailure(c, constants.RespCodeInvalidParams, err.Error(), nil)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
result := dataObjectAttributeQueryResult{
|
logger.Error(ctx, "build measurement attribute value failed", "token", token, "field", field, "error", err)
|
||||||
Token: token,
|
renderRespFailure(c, constants.RespCodeServerError, "query measurement attribute failed", nil)
|
||||||
Field: field,
|
return
|
||||||
Code: errcode.ErrProcessSuccess.Code(),
|
}
|
||||||
Msg: errcode.ErrProcessSuccess.Msg(),
|
|
||||||
Value: value,
|
result := dataObjectAttributeQueryResult{
|
||||||
|
Token: token,
|
||||||
|
Field: field,
|
||||||
|
Code: errcode.ErrProcessSuccess.Code(),
|
||||||
|
Msg: errcode.ErrProcessSuccess.Msg(),
|
||||||
|
Value: value,
|
||||||
|
}
|
||||||
|
renderRespSuccess(c, constants.RespCodeSuccess, "query measurement attribute success", map[string]any{
|
||||||
|
"attributes": []dataObjectAttributeQueryResult{result},
|
||||||
|
})
|
||||||
|
default:
|
||||||
|
renderRespFailure(c, constants.RespCodeInvalidParams, "invalid data object type", nil)
|
||||||
}
|
}
|
||||||
renderRespSuccess(c, constants.RespCodeSuccess, dataObjectAttributeSuccessMessage(dataObjectType), map[string]any{
|
|
||||||
"attributes": []dataObjectAttributeQueryResult{result},
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func parseDataObjectAttributeQuery(c *gin.Context) (string, string, error) {
|
func parseDataObjectAttributeQuery(c *gin.Context) (string, string, error) {
|
||||||
|
|
@ -91,11 +165,11 @@ func parseDataObjectAttributeQuery(c *gin.Context) (string, string, error) {
|
||||||
return token, field, nil
|
return token, field, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type dataObjectHashFieldLoader func(context.Context, constants.DataObjectType, string, string) (string, error)
|
type measurementValueLoader func(context.Context, orm.JSONMap) (any, error)
|
||||||
|
|
||||||
type measurementValueMetadataLoader func(context.Context, string) (orm.JSONMap, int, error)
|
type parameterValueLoader func(context.Context, *database.ParameterDataObject) (any, error)
|
||||||
|
|
||||||
type measurementValueLoader func(context.Context, orm.JSONMap, int) (any, error)
|
type parameterDescriptionLoader func(context.Context, string) (string, error)
|
||||||
|
|
||||||
var measurementDataObjectFields = map[string]struct{}{
|
var measurementDataObjectFields = map[string]struct{}{
|
||||||
"value": {},
|
"value": {},
|
||||||
|
|
@ -146,227 +220,116 @@ func validateDataObjectField(dataObjectType constants.DataObjectType, field stri
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func queryDataObjectAttributeValue(
|
func buildParameterAttributeValue(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
dataObjectType constants.DataObjectType,
|
|
||||||
token string,
|
|
||||||
field string,
|
field string,
|
||||||
loadHashField dataObjectHashFieldLoader,
|
parameter *database.ParameterDataObject,
|
||||||
loadMeasurementMetadata measurementValueMetadataLoader,
|
loadValue parameterValueLoader,
|
||||||
loadMeasurementValue measurementValueLoader,
|
loadDescription parameterDescriptionLoader,
|
||||||
) (any, error) {
|
) (any, error) {
|
||||||
if dataObjectType == constants.DataObjectTypeMeasurement && field == "value" {
|
if parameter == nil {
|
||||||
if loadMeasurementMetadata == nil {
|
return nil, fmt.Errorf("parameter data object is nil")
|
||||||
return nil, fmt.Errorf("measurement value metadata loader is nil")
|
|
||||||
}
|
|
||||||
if loadMeasurementValue == nil {
|
|
||||||
return nil, fmt.Errorf("measurement value loader is nil")
|
|
||||||
}
|
|
||||||
dataSource, size, err := loadMeasurementMetadata(ctx, token)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return loadMeasurementValue(ctx, dataSource, size)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if loadHashField == nil {
|
component := parameter.Component
|
||||||
return nil, fmt.Errorf("data-object hash field loader is nil")
|
switch field {
|
||||||
}
|
case "value":
|
||||||
rawValue, err := loadHashField(ctx, dataObjectType, token, field)
|
if loadValue == nil {
|
||||||
if err != nil {
|
return nil, fmt.Errorf("parameter value loader is nil")
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
if dataObjectType == constants.DataObjectTypeParameter && field == "value" {
|
|
||||||
attributeType, err := loadHashField(ctx, dataObjectType, token, "type")
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
}
|
||||||
return decodeParameterHashValue(rawValue, attributeType)
|
return loadValue(ctx, parameter)
|
||||||
|
case "meta":
|
||||||
|
return "PARAM", nil
|
||||||
|
case "type":
|
||||||
|
return parameter.AttributeType, nil
|
||||||
|
case "name":
|
||||||
|
return strings.Join([]string{
|
||||||
|
component.NSPath,
|
||||||
|
component.Tag,
|
||||||
|
parameter.AttributeGroup,
|
||||||
|
parameter.AttributeName,
|
||||||
|
}, "."), nil
|
||||||
|
case "description":
|
||||||
|
if loadDescription == nil {
|
||||||
|
return nil, fmt.Errorf("parameter description loader is nil")
|
||||||
|
}
|
||||||
|
return loadDescription(ctx, parameter.AttributeName)
|
||||||
|
case "id":
|
||||||
|
return strings.Join([]string{
|
||||||
|
component.GridName,
|
||||||
|
component.ZoneName,
|
||||||
|
component.StationName,
|
||||||
|
component.NSPath,
|
||||||
|
component.Tag,
|
||||||
|
parameter.AttributeGroup,
|
||||||
|
parameter.AttributeName,
|
||||||
|
}, "."), nil
|
||||||
|
default:
|
||||||
|
return nil, fmt.Errorf("%w: %s", common.ErrUnsupportedParameterField, field)
|
||||||
}
|
}
|
||||||
return decodeDataObjectHashField(dataObjectType, field, rawValue)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func loadDataObjectHashField(
|
func buildMeasurementAttributeValue(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
dataObjectType constants.DataObjectType,
|
|
||||||
token string,
|
|
||||||
field string,
|
field string,
|
||||||
) (string, error) {
|
measurement *orm.Measurement,
|
||||||
rdb := diagram.GetRedisClientInstance()
|
component *orm.Component,
|
||||||
if rdb == nil {
|
loadValue measurementValueLoader,
|
||||||
return "", fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
canonicalKey, err := model.ResolveDataObjectRedisKey(ctx, rdb, dataObjectType, token)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
value, err := rdb.HGet(ctx, canonicalKey, field).Result()
|
|
||||||
if errors.Is(err, redis.Nil) {
|
|
||||||
return "", fmt.Errorf("canonical redis data-object hash %q does not contain field %q", canonicalKey, field)
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
return "", fmt.Errorf("query canonical redis hash %q field %q: %w", canonicalKey, field, err)
|
|
||||||
}
|
|
||||||
return value, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func loadMeasurementValueMetadata(ctx context.Context, token string) (orm.JSONMap, int, error) {
|
|
||||||
rdb := diagram.GetRedisClientInstance()
|
|
||||||
if rdb == nil {
|
|
||||||
return nil, 0, fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
canonicalKey, err := model.ResolveDataObjectRedisKey(ctx, rdb, constants.DataObjectTypeMeasurement, token)
|
|
||||||
if err != nil {
|
|
||||||
return nil, 0, err
|
|
||||||
}
|
|
||||||
values, err := rdb.HMGet(ctx, canonicalKey, "data_source", "size").Result()
|
|
||||||
if err != nil {
|
|
||||||
return nil, 0, fmt.Errorf("query canonical redis hash %q measurement value metadata: %w", canonicalKey, err)
|
|
||||||
}
|
|
||||||
if len(values) != 2 {
|
|
||||||
return nil, 0, fmt.Errorf("canonical redis hash %q returned %d measurement metadata fields", canonicalKey, len(values))
|
|
||||||
}
|
|
||||||
if values[0] == nil || values[1] == nil {
|
|
||||||
missingFields := make([]string, 0, 2)
|
|
||||||
if values[0] == nil {
|
|
||||||
missingFields = append(missingFields, "data_source")
|
|
||||||
}
|
|
||||||
if values[1] == nil {
|
|
||||||
missingFields = append(missingFields, "size")
|
|
||||||
}
|
|
||||||
return nil, 0, fmt.Errorf(
|
|
||||||
"canonical redis measurement hash %q does not contain field(s) %s",
|
|
||||||
canonicalKey,
|
|
||||||
strings.Join(missingFields, ", "),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
rawDataSource, ok := values[0].(string)
|
|
||||||
if !ok {
|
|
||||||
return nil, 0, fmt.Errorf("canonical redis measurement hash %q data_source has type %T", canonicalKey, values[0])
|
|
||||||
}
|
|
||||||
var dataSource orm.JSONMap
|
|
||||||
if err := json.Unmarshal([]byte(rawDataSource), &dataSource); err != nil {
|
|
||||||
return nil, 0, fmt.Errorf("decode measurement data_source from canonical redis hash %q: %w", canonicalKey, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
rawSize, ok := values[1].(string)
|
|
||||||
if !ok {
|
|
||||||
return nil, 0, fmt.Errorf("canonical redis measurement hash %q size has type %T", canonicalKey, values[1])
|
|
||||||
}
|
|
||||||
size, err := strconv.Atoi(rawSize)
|
|
||||||
if err != nil {
|
|
||||||
return nil, 0, fmt.Errorf("decode measurement size %q: %w", rawSize, err)
|
|
||||||
}
|
|
||||||
if size <= 0 {
|
|
||||||
return nil, 0, fmt.Errorf("measurement window size must be greater than 0, got %d", size)
|
|
||||||
}
|
|
||||||
return dataSource, size, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func decodeDataObjectHashField(
|
|
||||||
dataObjectType constants.DataObjectType,
|
|
||||||
field string,
|
|
||||||
rawValue string,
|
|
||||||
) (any, error) {
|
) (any, error) {
|
||||||
if dataObjectType != constants.DataObjectTypeMeasurement {
|
if measurement == nil {
|
||||||
return rawValue, nil
|
return nil, fmt.Errorf("measurement is nil")
|
||||||
|
}
|
||||||
|
if component == nil {
|
||||||
|
return nil, fmt.Errorf("measurement component is nil")
|
||||||
}
|
}
|
||||||
|
|
||||||
switch field {
|
switch field {
|
||||||
|
case "value":
|
||||||
|
if loadValue == nil {
|
||||||
|
return nil, fmt.Errorf("measurement value loader is nil")
|
||||||
|
}
|
||||||
|
return loadValue(ctx, measurement.DataSource)
|
||||||
case "mode":
|
case "mode":
|
||||||
value, err := strconv.ParseInt(rawValue, 10, 16)
|
return measurement.Mode, nil
|
||||||
if err != nil {
|
case "meta":
|
||||||
return nil, fmt.Errorf("decode measurement mode %q: %w", rawValue, err)
|
return "MEASUREMENT", nil
|
||||||
}
|
case "type":
|
||||||
return int16(value), nil
|
return model.MeasurementTypeFromDataSource(measurement.DataSource)
|
||||||
|
case "name":
|
||||||
|
// The resolved measurement and component prove that token4.token7 exists.
|
||||||
|
return component.NSPath + "." + measurement.Tag, nil
|
||||||
|
case "description":
|
||||||
|
return measurement.Name, nil
|
||||||
|
case "id":
|
||||||
|
return strings.Join([]string{
|
||||||
|
component.GridName,
|
||||||
|
component.ZoneName,
|
||||||
|
component.StationName,
|
||||||
|
component.NSPath,
|
||||||
|
component.Tag,
|
||||||
|
"bay",
|
||||||
|
measurement.Tag,
|
||||||
|
}, "."), nil
|
||||||
case "size":
|
case "size":
|
||||||
value, err := strconv.Atoi(rawValue)
|
return measurement.Size, nil
|
||||||
if err != nil {
|
case "data_source":
|
||||||
return nil, fmt.Errorf("decode measurement size %q: %w", rawValue, err)
|
return measurement.DataSource, nil
|
||||||
}
|
case "event_plan":
|
||||||
return value, nil
|
return measurement.EventPlan, nil
|
||||||
case "data_source", "event_plan", "binding":
|
case "binding":
|
||||||
return decodeRedisJSON(rawValue)
|
return measurement.Binding, nil
|
||||||
default:
|
default:
|
||||||
return rawValue, nil
|
return nil, fmt.Errorf("%w: %s", common.ErrUnsupportedMeasurementField, field)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func decodeParameterHashValue(rawValue, attributeType string) (any, error) {
|
func queryMeasurementRealtimeValue(ctx context.Context, dataSource orm.JSONMap) (any, error) {
|
||||||
normalizedType := strings.ToUpper(strings.TrimSpace(attributeType))
|
|
||||||
if rawValue == "null" {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
switch {
|
|
||||||
case normalizedType == "BOOLEAN":
|
|
||||||
value, err := strconv.ParseBool(rawValue)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("decode parameter boolean value %q: %w", rawValue, err)
|
|
||||||
}
|
|
||||||
return value, nil
|
|
||||||
case normalizedType == "SMALLINT",
|
|
||||||
normalizedType == "INTEGER",
|
|
||||||
normalizedType == "BIGINT":
|
|
||||||
value, err := strconv.ParseInt(rawValue, 10, 64)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("decode parameter integer value %q: %w", rawValue, err)
|
|
||||||
}
|
|
||||||
return value, nil
|
|
||||||
case normalizedType == "REAL",
|
|
||||||
normalizedType == "DOUBLE PRECISION",
|
|
||||||
strings.HasPrefix(normalizedType, "NUMERIC"),
|
|
||||||
strings.HasPrefix(normalizedType, "DECIMAL"):
|
|
||||||
if _, err := strconv.ParseFloat(rawValue, 64); err != nil {
|
|
||||||
return nil, fmt.Errorf("decode parameter numeric value %q: %w", rawValue, err)
|
|
||||||
}
|
|
||||||
return json.Number(rawValue), nil
|
|
||||||
case normalizedType == "JSON",
|
|
||||||
normalizedType == "JSONB",
|
|
||||||
strings.HasSuffix(normalizedType, "[]"):
|
|
||||||
return decodeRedisJSON(rawValue)
|
|
||||||
default:
|
|
||||||
return rawValue, nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func decodeRedisJSON(rawValue string) (any, error) {
|
|
||||||
var value any
|
|
||||||
decoder := json.NewDecoder(strings.NewReader(rawValue))
|
|
||||||
decoder.UseNumber()
|
|
||||||
if err := decoder.Decode(&value); err != nil {
|
|
||||||
return nil, fmt.Errorf("decode redis JSON value: %w", err)
|
|
||||||
}
|
|
||||||
return value, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func isDataObjectTokenNotFound(err error) bool {
|
|
||||||
return errors.Is(err, common.ErrParameterTokenNotFound) ||
|
|
||||||
errors.Is(err, common.ErrMeasurementTokenNotFound)
|
|
||||||
}
|
|
||||||
|
|
||||||
func dataObjectAttributeSuccessMessage(dataObjectType constants.DataObjectType) string {
|
|
||||||
if dataObjectType == constants.DataObjectTypeParameter {
|
|
||||||
return "query parameter attribute success"
|
|
||||||
}
|
|
||||||
return "query measurement attribute success"
|
|
||||||
}
|
|
||||||
|
|
||||||
func dataObjectAttributeFailureMessage(dataObjectType constants.DataObjectType) string {
|
|
||||||
if dataObjectType == constants.DataObjectTypeParameter {
|
|
||||||
return "query parameter attribute failed"
|
|
||||||
}
|
|
||||||
return "query measurement attribute failed"
|
|
||||||
}
|
|
||||||
|
|
||||||
func queryMeasurementRealtimeValue(ctx context.Context, dataSource orm.JSONMap, size int) (any, error) {
|
|
||||||
queryKey, err := model.GenerateMeasureIdentifier(dataSource)
|
queryKey, err := model.GenerateMeasureIdentifier(dataSource)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("generate measurement redis key: %w", err)
|
return nil, fmt.Errorf("generate measurement redis key: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
value, err := diagram.NewRedisClient().QueryLatestMeasurementValues(ctx, queryKey, size)
|
value, err := diagram.NewRedisClient().QueryLatestMeasurementValue(ctx, queryKey)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("query real-time measurement value by key %q: %w", queryKey, err)
|
return nil, fmt.Errorf("query real-time measurement value by key %q: %w", queryKey, err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -2,14 +2,13 @@ package handler
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"modelRT/common"
|
"modelRT/common"
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
|
"modelRT/database"
|
||||||
"modelRT/orm"
|
"modelRT/orm"
|
||||||
|
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
|
|
@ -118,156 +117,160 @@ func TestValidateDataObjectField(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestQueryParameterAttributeValueFromRedisHash(t *testing.T) {
|
func TestBuildParameterAttributeValue(t *testing.T) {
|
||||||
fields := map[string]string{
|
parameter := &database.ParameterDataObject{
|
||||||
"value": "220.50",
|
Component: orm.Component{
|
||||||
"type": "DOUBLE PRECISION",
|
GridName: "grid000",
|
||||||
"name": "110kV_TV.cable_22.rated.rated_voltage",
|
ZoneName: "zone000",
|
||||||
"description": "额定电压",
|
StationName: "station000",
|
||||||
}
|
NSPath: "110kV_TV",
|
||||||
loader := hashFieldLoaderForTest(fields)
|
Tag: "cable_22",
|
||||||
|
|
||||||
value, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeParameter,
|
|
||||||
"parameter-token",
|
|
||||||
"value",
|
|
||||||
loader,
|
|
||||||
nil,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, json.Number("220.50"), value)
|
|
||||||
|
|
||||||
description, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeParameter,
|
|
||||||
"parameter-token",
|
|
||||||
"description",
|
|
||||||
loader,
|
|
||||||
nil,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "额定电压", description)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestQueryMeasurementAttributeValueFromRedisHash(t *testing.T) {
|
|
||||||
fields := map[string]string{
|
|
||||||
"mode": "1",
|
|
||||||
"size": "10",
|
|
||||||
"name": "110kV_TV.IA_rms",
|
|
||||||
"data_source": `{"type":1,"io_address":{"channel":"tm1p"}}`,
|
|
||||||
"event_plan": `{"enabled":true}`,
|
|
||||||
}
|
|
||||||
loader := hashFieldLoaderForTest(fields)
|
|
||||||
|
|
||||||
mode, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeMeasurement,
|
|
||||||
"measurement-token",
|
|
||||||
"mode",
|
|
||||||
loader,
|
|
||||||
nil,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, int16(1), mode)
|
|
||||||
|
|
||||||
eventPlan, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeMeasurement,
|
|
||||||
"measurement-token",
|
|
||||||
"event_plan",
|
|
||||||
loader,
|
|
||||||
nil,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, map[string]any{"enabled": true}, eventPlan)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestQueryMeasurementRealtimeValueUsesDataSourceFromRedisHash(t *testing.T) {
|
|
||||||
dataSource := orm.JSONMap{
|
|
||||||
"type": float64(1),
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"station": "001",
|
|
||||||
"channel": "tm1p",
|
|
||||||
},
|
},
|
||||||
|
AttributeGroup: "rated",
|
||||||
|
AttributeName: "rated_voltage",
|
||||||
|
AttributeType: "DOUBLE PRECISION",
|
||||||
}
|
}
|
||||||
metadataLoader := func(_ context.Context, token string) (orm.JSONMap, int, error) {
|
loader := func(_ context.Context, actual *database.ParameterDataObject) (any, error) {
|
||||||
assert.Equal(t, "measurement-token", token)
|
assert.Same(t, parameter, actual)
|
||||||
return dataSource, 2, nil
|
return float64(220), nil
|
||||||
}
|
}
|
||||||
valueLoader := func(_ context.Context, dataSource orm.JSONMap, size int) (any, error) {
|
descriptionLoader := func(_ context.Context, attributeName string) (string, error) {
|
||||||
assert.Equal(t, float64(1), dataSource["type"])
|
assert.Equal(t, "rated_voltage", attributeName)
|
||||||
assert.Equal(t, "001", dataSource["io_address"].(map[string]any)["station"])
|
return "额定电压", nil
|
||||||
assert.Equal(t, 2, size)
|
|
||||||
return []float64{220, 219.5}, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
value, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeMeasurement,
|
|
||||||
"measurement-token",
|
|
||||||
"value",
|
|
||||||
nil,
|
|
||||||
metadataLoader,
|
|
||||||
valueLoader,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, []float64{220, 219.5}, value)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestQueryDataObjectAttributeValuePropagatesRedisTokenNotFound(t *testing.T) {
|
|
||||||
loader := func(context.Context, constants.DataObjectType, string, string) (string, error) {
|
|
||||||
return "", fmt.Errorf("%w: token", common.ErrParameterTokenNotFound)
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err := queryDataObjectAttributeValue(
|
|
||||||
context.Background(),
|
|
||||||
constants.DataObjectTypeParameter,
|
|
||||||
"missing-token",
|
|
||||||
"name",
|
|
||||||
loader,
|
|
||||||
nil,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.ErrorIs(t, err, common.ErrParameterTokenNotFound)
|
|
||||||
assert.True(t, isDataObjectTokenNotFound(err))
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDecodeParameterHashValue(t *testing.T) {
|
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
field string
|
||||||
rawValue string
|
expected any
|
||||||
attributeType string
|
|
||||||
expected any
|
|
||||||
}{
|
}{
|
||||||
{name: "boolean", rawValue: "true", attributeType: "BOOLEAN", expected: true},
|
{field: "value", expected: float64(220)},
|
||||||
{name: "integer", rawValue: "42", attributeType: "INTEGER", expected: int64(42)},
|
{field: "meta", expected: "PARAM"},
|
||||||
{name: "numeric", rawValue: "1234567890.123456789", attributeType: "NUMERIC(30,9)", expected: json.Number("1234567890.123456789")},
|
{field: "type", expected: "DOUBLE PRECISION"},
|
||||||
{name: "jsonb", rawValue: `{"key":"value"}`, attributeType: "JSONB", expected: map[string]any{"key": "value"}},
|
{field: "name", expected: "110kV_TV.cable_22.rated.rated_voltage"},
|
||||||
{name: "string", rawValue: "cable", attributeType: "CHARACTER VARYING(64)", expected: "cable"},
|
{field: "description", expected: "额定电压"},
|
||||||
{name: "null", rawValue: "null", attributeType: "INTEGER", expected: nil},
|
{field: "id", expected: "grid000.zone000.station000.110kV_TV.cable_22.rated.rated_voltage"},
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, test := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(test.name, func(t *testing.T) {
|
t.Run(tt.field, func(t *testing.T) {
|
||||||
actual, err := decodeParameterHashValue(test.rawValue, test.attributeType)
|
actual, err := buildParameterAttributeValue(
|
||||||
|
context.Background(),
|
||||||
|
tt.field,
|
||||||
|
parameter,
|
||||||
|
loader,
|
||||||
|
descriptionLoader,
|
||||||
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, test.expected, actual)
|
assert.Equal(t, tt.expected, actual)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func hashFieldLoaderForTest(fields map[string]string) dataObjectHashFieldLoader {
|
func TestBuildParameterAttributeValueRejectsUnsupportedField(t *testing.T) {
|
||||||
return func(_ context.Context, _ constants.DataObjectType, _ string, field string) (string, error) {
|
_, err := buildParameterAttributeValue(
|
||||||
value, exists := fields[field]
|
context.Background(),
|
||||||
if !exists {
|
"unknown",
|
||||||
return "", fmt.Errorf("field %q not found", field)
|
&database.ParameterDataObject{},
|
||||||
}
|
nil,
|
||||||
return value, nil
|
nil,
|
||||||
|
)
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.ErrorIs(t, err, common.ErrUnsupportedParameterField)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildMeasurementAttributeValue(t *testing.T) {
|
||||||
|
dataSource := orm.JSONMap{
|
||||||
|
"type": float64(1),
|
||||||
|
"io_address": map[string]any{
|
||||||
|
"channel": "tm1p",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
eventPlan := orm.JSONMap{"enabled": true}
|
||||||
|
binding := orm.JSONMap{"ct": map[string]any{"ratio": float64(2)}}
|
||||||
|
measurement := &orm.Measurement{
|
||||||
|
Tag: "IA_rms",
|
||||||
|
Name: "A相电流",
|
||||||
|
Mode: 1,
|
||||||
|
Size: 10,
|
||||||
|
DataSource: dataSource,
|
||||||
|
EventPlan: eventPlan,
|
||||||
|
Binding: binding,
|
||||||
|
}
|
||||||
|
component := &orm.Component{
|
||||||
|
GridName: "grid000",
|
||||||
|
ZoneName: "zone000",
|
||||||
|
StationName: "station000",
|
||||||
|
NSPath: "110kV_TV",
|
||||||
|
Tag: "cable_22",
|
||||||
|
}
|
||||||
|
|
||||||
|
loader := func(_ context.Context, source orm.JSONMap) (any, error) {
|
||||||
|
assert.Equal(t, dataSource, source)
|
||||||
|
return float64(220), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
field string
|
||||||
|
expected any
|
||||||
|
}{
|
||||||
|
{field: "value", expected: float64(220)},
|
||||||
|
{field: "mode", expected: int16(1)},
|
||||||
|
{field: "meta", expected: "MEASUREMENT"},
|
||||||
|
{field: "type", expected: "TM"},
|
||||||
|
{field: "name", expected: "110kV_TV.IA_rms"},
|
||||||
|
{field: "description", expected: "A相电流"},
|
||||||
|
{field: "id", expected: "grid000.zone000.station000.110kV_TV.cable_22.bay.IA_rms"},
|
||||||
|
{field: "size", expected: 10},
|
||||||
|
{field: "data_source", expected: dataSource},
|
||||||
|
{field: "event_plan", expected: eventPlan},
|
||||||
|
{field: "binding", expected: binding},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.field, func(t *testing.T) {
|
||||||
|
actual, err := buildMeasurementAttributeValue(context.Background(), tt.field, measurement, component, loader)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, tt.expected, actual)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildMeasurementAttributeValueRejectsUnsupportedField(t *testing.T) {
|
||||||
|
_, err := buildMeasurementAttributeValue(
|
||||||
|
context.Background(),
|
||||||
|
"unknown",
|
||||||
|
&orm.Measurement{},
|
||||||
|
&orm.Component{},
|
||||||
|
nil,
|
||||||
|
)
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.ErrorIs(t, err, common.ErrUnsupportedMeasurementField)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildMeasurementAttributeValueMode(t *testing.T) {
|
||||||
|
component := &orm.Component{}
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
mode int16
|
||||||
|
expected int16
|
||||||
|
}{
|
||||||
|
{name: "collected value", mode: 1, expected: 1},
|
||||||
|
{name: "manually assigned value", mode: 0, expected: 0},
|
||||||
|
{name: "other positive mode", mode: 2, expected: 2},
|
||||||
|
{name: "negative mode", mode: -1, expected: -1},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
actual, err := buildMeasurementAttributeValue(
|
||||||
|
context.Background(),
|
||||||
|
"mode",
|
||||||
|
&orm.Measurement{Mode: tt.mode},
|
||||||
|
component,
|
||||||
|
nil,
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, tt.expected, actual)
|
||||||
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,6 @@ import (
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"modelRT/client/manualsync"
|
|
||||||
"modelRT/common"
|
"modelRT/common"
|
||||||
"modelRT/common/errcode"
|
"modelRT/common/errcode"
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
|
|
@ -20,7 +19,6 @@ import (
|
||||||
"modelRT/logger"
|
"modelRT/logger"
|
||||||
"modelRT/model"
|
"modelRT/model"
|
||||||
"modelRT/orm"
|
"modelRT/orm"
|
||||||
redisrepository "modelRT/repository/redis"
|
|
||||||
|
|
||||||
"github.com/gin-gonic/gin"
|
"github.com/gin-gonic/gin"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
|
|
@ -33,8 +31,6 @@ type dataObjectAttributeUpdateRequest struct {
|
||||||
Data json.RawMessage `json:"data,omitempty"`
|
Data json.RawMessage `json:"data,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
const redisChangeRestoreTimeout = 5 * time.Second
|
|
||||||
|
|
||||||
// DataObjectAttributeUpdateHandler updates the writable field of one data object.
|
// DataObjectAttributeUpdateHandler updates the writable field of one data object.
|
||||||
func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
||||||
ctx := c.Request.Context()
|
ctx := c.Request.Context()
|
||||||
|
|
@ -65,50 +61,22 @@ func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
redisClient := diagram.GetRedisClientInstance()
|
|
||||||
redisChanges := redisrepository.NewRedisChangeSet(redisClient)
|
|
||||||
canonicalRedisKey, err := model.ResolveDataObjectRedisKey(
|
|
||||||
ctx,
|
|
||||||
redisClient,
|
|
||||||
dataObjectType,
|
|
||||||
request.Token,
|
|
||||||
)
|
|
||||||
message := "data-object attribute update success"
|
message := "data-object attribute update success"
|
||||||
var measurementResult measurementUpdateResult
|
var measurementResult measurementUpdateResult
|
||||||
switch {
|
switch dataObjectType {
|
||||||
case err != nil:
|
case constants.DataObjectTypeParameter:
|
||||||
// The shared resolver error is handled by the common failure path below.
|
|
||||||
case dataObjectType == constants.DataObjectTypeParameter:
|
|
||||||
parameter, queryErr := database.QueryParameterByDataObjectToken(ctx, tx, request.Token)
|
parameter, queryErr := database.QueryParameterByDataObjectToken(ctx, tx, request.Token)
|
||||||
if queryErr == nil {
|
if queryErr == nil {
|
||||||
queryErr = database.UpdateParameterDataObjectValue(ctx, tx, parameter, value)
|
queryErr = database.UpdateParameterDataObjectValue(ctx, tx, parameter, value)
|
||||||
}
|
}
|
||||||
if queryErr == nil {
|
|
||||||
queryErr = redisChanges.AddHashChange(ctx, canonicalRedisKey, field, value)
|
|
||||||
}
|
|
||||||
err = queryErr
|
err = queryErr
|
||||||
case dataObjectType == constants.DataObjectTypeMeasurement:
|
case constants.DataObjectTypeMeasurement:
|
||||||
measurementResult, err = updateMeasurementDataObject(ctx, tx, request.Token, field, value, request.Data, measurementUpdateDependencies{
|
measurementResult, err = updateMeasurementDataObject(ctx, tx, request.Token, field, value, request.Data, measurementUpdateDependencies{
|
||||||
writeManualValueFunc: func(ctx context.Context, measurement *orm.Measurement, value float64, timestamp time.Time) error {
|
writeManualValueFunc: writeMeasurementManualValue,
|
||||||
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
updateDataRTFunc: callRealTimeDataWriteStopInterface,
|
||||||
if err != nil {
|
startDataRTFunc: callRealTimeDataWriteStartInterface,
|
||||||
return fmt.Errorf("generate measurement redis key: %w", err)
|
replaceRedisValueFunc: replaceMeasurementRedisValue,
|
||||||
}
|
|
||||||
return redisChanges.AddMeasurementValueChange(ctx, key, value, timestamp, false)
|
|
||||||
},
|
|
||||||
syncManualChangeFunc: manualsync.Sync,
|
|
||||||
replaceRedisValueFunc: func(ctx context.Context, measurement *orm.Measurement, value float64, timestamp time.Time) error {
|
|
||||||
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("generate measurement redis key: %w", err)
|
|
||||||
}
|
|
||||||
return redisChanges.AddMeasurementValueChange(ctx, key, value, timestamp, true)
|
|
||||||
},
|
|
||||||
nowFunc: time.Now,
|
|
||||||
})
|
})
|
||||||
if err == nil && measurementResult.modeChanged {
|
|
||||||
err = redisChanges.AddHashChange(ctx, canonicalRedisKey, "mode", measurementResult.mode)
|
|
||||||
}
|
|
||||||
message = measurementResult.message
|
message = measurementResult.message
|
||||||
default:
|
default:
|
||||||
err = fmt.Errorf("unsupported data object type %q", dataObjectType)
|
err = fmt.Errorf("unsupported data object type %q", dataObjectType)
|
||||||
|
|
@ -116,8 +84,8 @@ func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_ = tx.Rollback().Error
|
_ = tx.Rollback().Error
|
||||||
if measurementResult.recordFailureOnError {
|
if measurementResult.recordFailure {
|
||||||
if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); logErr != nil {
|
if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, time.Now().UTC()); logErr != nil {
|
||||||
logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
|
logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -130,29 +98,7 @@ func DataObjectAttributeUpdateHandler(c *gin.Context) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := redisChanges.Apply(ctx); err != nil {
|
|
||||||
_ = tx.Rollback().Error
|
|
||||||
if measurementResult.recordFailureOnError {
|
|
||||||
if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); logErr != nil {
|
|
||||||
logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
logger.Error(ctx, "apply redis data-object changes failed", "token", request.Token, "field", field, "error", err)
|
|
||||||
renderRespFailure(c, constants.RespCodeFailed, err.Error(), nil)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := tx.Commit().Error; err != nil {
|
if err := tx.Commit().Error; err != nil {
|
||||||
revertCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), redisChangeRestoreTimeout)
|
|
||||||
defer cancel()
|
|
||||||
if redisErr := redisChanges.Revert(revertCtx); redisErr != nil {
|
|
||||||
logger.Error(ctx, "revert redis data-object changes failed", "token", request.Token, "field", field, "error", redisErr)
|
|
||||||
}
|
|
||||||
if measurementResult.recordFailureOnError {
|
|
||||||
if logErr := database.AppendMeasurementValueOperation(ctx, database.GetPostgresDBClient(), measurementResult.measurementID, 1, measurementResult.value, measurementFailureTime(measurementResult)); logErr != nil {
|
|
||||||
logger.Error(ctx, "append failed measurement value operation failed", "measurement_id", measurementResult.measurementID, "error", logErr)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
logger.Error(ctx, "commit data-object update transaction failed", "token", request.Token, "field", field, "error", err)
|
logger.Error(ctx, "commit data-object update transaction failed", "token", request.Token, "field", field, "error", err)
|
||||||
renderRespFailure(c, constants.RespCodeServerError, "transaction commit failed", nil)
|
renderRespFailure(c, constants.RespCodeServerError, "transaction commit failed", nil)
|
||||||
return
|
return
|
||||||
|
|
@ -273,27 +219,24 @@ func parseMeasurementUpdateMode(raw json.RawMessage) (int16, error) {
|
||||||
return mode, nil
|
return mode, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type measurementManualValueWriter func(context.Context, *orm.Measurement, float64, time.Time) error
|
type measurementManualValueWriter func(context.Context, *orm.Measurement, float64) error
|
||||||
|
|
||||||
type measurementManualChangeSyncer func(context.Context, orm.JSONMap, int16, *manualsync.SyntheticData) error
|
type measurementDataRTUpdater func(context.Context, orm.JSONMap, *float64) error
|
||||||
|
|
||||||
type measurementRedisValueReplacer func(context.Context, *orm.Measurement, float64, time.Time) error
|
type measurementRedisValueReplacer func(context.Context, *orm.Measurement, float64) error
|
||||||
|
|
||||||
type measurementUpdateDependencies struct {
|
type measurementUpdateDependencies struct {
|
||||||
writeManualValueFunc measurementManualValueWriter
|
writeManualValueFunc measurementManualValueWriter
|
||||||
syncManualChangeFunc measurementManualChangeSyncer
|
updateDataRTFunc measurementDataRTUpdater
|
||||||
|
startDataRTFunc measurementDataRTUpdater
|
||||||
replaceRedisValueFunc measurementRedisValueReplacer
|
replaceRedisValueFunc measurementRedisValueReplacer
|
||||||
nowFunc func() time.Time
|
|
||||||
}
|
}
|
||||||
|
|
||||||
type measurementUpdateResult struct {
|
type measurementUpdateResult struct {
|
||||||
message string
|
message string
|
||||||
measurementID int64
|
measurementID int64
|
||||||
value float64
|
value float64
|
||||||
recordFailureOnError bool
|
recordFailure bool
|
||||||
mode int16
|
|
||||||
modeChanged bool
|
|
||||||
operationTime time.Time
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func updateMeasurementDataObject(
|
func updateMeasurementDataObject(
|
||||||
|
|
@ -328,7 +271,6 @@ func updateMeasurementDataObject(
|
||||||
if currentMode == targetAutomatic {
|
if currentMode == targetAutomatic {
|
||||||
return measurementUpdateResult{message: fmt.Sprintf("measurement is already in %s mode", measurementModeName(mode))}, nil
|
return measurementUpdateResult{message: fmt.Sprintf("measurement is already in %s mode", measurementModeName(mode))}, nil
|
||||||
}
|
}
|
||||||
operationTime := measurementUpdateNow(dependencies)
|
|
||||||
var manualValue *float64
|
var manualValue *float64
|
||||||
if currentMode && mode == constants.MeasurementModeManual {
|
if currentMode && mode == constants.MeasurementModeManual {
|
||||||
manualValue, err = parseOptionalMeasurementModeData(modeData)
|
manualValue, err = parseOptionalMeasurementModeData(modeData)
|
||||||
|
|
@ -336,33 +278,34 @@ func updateMeasurementDataObject(
|
||||||
return measurementUpdateResult{}, err
|
return measurementUpdateResult{}, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := database.UpdateMeasurementModeWithOperation(ctx, tx, lockedMeasurement.ID, mode, operationTime); err != nil {
|
if err := database.UpdateMeasurementModeWithOperation(ctx, tx, lockedMeasurement.ID, mode, time.Now().UTC()); err != nil {
|
||||||
return measurementUpdateResult{}, err
|
return measurementUpdateResult{}, err
|
||||||
}
|
}
|
||||||
syncManualMeasurementChange(
|
|
||||||
ctx,
|
|
||||||
dependencies.syncManualChangeFunc,
|
|
||||||
lockedMeasurement.ID,
|
|
||||||
lockedMeasurement.DataSource,
|
|
||||||
mode,
|
|
||||||
nil,
|
|
||||||
)
|
|
||||||
if currentMode && mode == constants.MeasurementModeManual {
|
if currentMode && mode == constants.MeasurementModeManual {
|
||||||
|
if dependencies.updateDataRTFunc == nil {
|
||||||
|
return measurementUpdateResult{}, fmt.Errorf("measurement dataRT updater is nil")
|
||||||
|
}
|
||||||
|
if err := dependencies.updateDataRTFunc(ctx, lockedMeasurement.DataSource, nil); err != nil {
|
||||||
|
return measurementUpdateResult{}, fmt.Errorf("stop automatic measurement write to dataRT: %w", err)
|
||||||
|
}
|
||||||
if manualValue != nil {
|
if manualValue != nil {
|
||||||
if dependencies.replaceRedisValueFunc == nil {
|
if dependencies.replaceRedisValueFunc == nil {
|
||||||
return measurementUpdateResult{}, fmt.Errorf("measurement redis value replacer is nil")
|
return measurementUpdateResult{}, fmt.Errorf("measurement redis value replacer is nil")
|
||||||
}
|
}
|
||||||
if err := dependencies.replaceRedisValueFunc(ctx, &lockedMeasurement, *manualValue, operationTime); err != nil {
|
if err := dependencies.replaceRedisValueFunc(ctx, &lockedMeasurement, *manualValue); err != nil {
|
||||||
return measurementUpdateResult{}, fmt.Errorf("replace measurement redis value: %w", err)
|
return measurementUpdateResult{}, fmt.Errorf("replace measurement redis value: %w", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return measurementUpdateResult{
|
if !currentMode && mode == constants.MeasurementModeAutomatic {
|
||||||
message: fmt.Sprintf("measurement mode changed to %s", measurementModeName(mode)),
|
if dependencies.startDataRTFunc == nil {
|
||||||
mode: mode,
|
return measurementUpdateResult{}, fmt.Errorf("measurement dataRT starter is nil")
|
||||||
modeChanged: true,
|
}
|
||||||
operationTime: operationTime,
|
if err := dependencies.startDataRTFunc(ctx, lockedMeasurement.DataSource, nil); err != nil {
|
||||||
}, nil
|
return measurementUpdateResult{}, fmt.Errorf("start automatic measurement write to dataRT: %w", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return measurementUpdateResult{message: fmt.Sprintf("measurement mode changed to %s", measurementModeName(mode))}, nil
|
||||||
case "value":
|
case "value":
|
||||||
currentMode, err := measurementModeIsAutomatic(lockedMeasurement.Mode)
|
currentMode, err := measurementModeIsAutomatic(lockedMeasurement.Mode)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -375,94 +318,32 @@ func updateMeasurementDataObject(
|
||||||
if !ok {
|
if !ok {
|
||||||
return measurementUpdateResult{}, fmt.Errorf("measurement value has invalid type %T", value)
|
return measurementUpdateResult{}, fmt.Errorf("measurement value has invalid type %T", value)
|
||||||
}
|
}
|
||||||
operationTime := measurementUpdateNow(dependencies)
|
|
||||||
failureResult := measurementUpdateResult{
|
failureResult := measurementUpdateResult{
|
||||||
measurementID: lockedMeasurement.ID,
|
measurementID: lockedMeasurement.ID,
|
||||||
value: manualValue,
|
value: manualValue,
|
||||||
recordFailureOnError: true,
|
recordFailure: true,
|
||||||
operationTime: operationTime,
|
|
||||||
}
|
}
|
||||||
if dependencies.writeManualValueFunc == nil {
|
if dependencies.writeManualValueFunc == nil {
|
||||||
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement manual value writer is nil"))
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement manual value writer is nil"))
|
||||||
}
|
}
|
||||||
if err := dependencies.writeManualValueFunc(ctx, &lockedMeasurement, manualValue, operationTime); err != nil {
|
if err := dependencies.writeManualValueFunc(ctx, &lockedMeasurement, manualValue); err != nil {
|
||||||
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
||||||
}
|
}
|
||||||
sample := manualsync.SyntheticData{Time: operationTime.UnixNano(), Value: manualValue}
|
if dependencies.updateDataRTFunc == nil {
|
||||||
syncManualMeasurementChange(
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(fmt.Errorf("measurement dataRT updater is nil"))
|
||||||
ctx,
|
}
|
||||||
dependencies.syncManualChangeFunc,
|
if err := dependencies.updateDataRTFunc(ctx, lockedMeasurement.DataSource, &manualValue); err != nil {
|
||||||
lockedMeasurement.ID,
|
|
||||||
lockedMeasurement.DataSource,
|
|
||||||
constants.MeasurementModeManual,
|
|
||||||
&sample,
|
|
||||||
)
|
|
||||||
if err := database.AppendMeasurementValueOperation(ctx, tx, lockedMeasurement.ID, 0, manualValue, operationTime); err != nil {
|
|
||||||
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
||||||
}
|
}
|
||||||
failureResult.message = "measurement manual value updated"
|
if err := database.AppendMeasurementValueOperation(ctx, tx, lockedMeasurement.ID, 0, manualValue, time.Now().UTC()); err != nil {
|
||||||
return failureResult, nil
|
return failureResult, errcode.ErrMeasurementValueUpdateFailed.WithCause(err)
|
||||||
|
}
|
||||||
|
return measurementUpdateResult{message: "measurement manual value updated"}, nil
|
||||||
default:
|
default:
|
||||||
return measurementUpdateResult{}, fmt.Errorf("unsupported measurement update field %q", field)
|
return measurementUpdateResult{}, fmt.Errorf("unsupported measurement update field %q", field)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO: This synchronous best-effort manual synchronization is not necessarily
|
|
||||||
// the final implementation. Its current behavior is to keep the local update
|
|
||||||
// successful when the downstream request fails (or the client is unavailable),
|
|
||||||
// log the error, and permanently drop that synchronization event without retry.
|
|
||||||
// The HTTP timeout still adds latency to the update request, and downstream state
|
|
||||||
// may diverge from local state. Revisit asynchronous delivery or a durable outbox
|
|
||||||
// if delivery reliability or request latency becomes important.
|
|
||||||
func syncManualMeasurementChange(
|
|
||||||
ctx context.Context,
|
|
||||||
syncer measurementManualChangeSyncer,
|
|
||||||
measurementID int64,
|
|
||||||
dataSource orm.JSONMap,
|
|
||||||
mode int16,
|
|
||||||
sample *manualsync.SyntheticData,
|
|
||||||
) {
|
|
||||||
if syncer == nil {
|
|
||||||
logManualMeasurementSyncError(ctx, "manual measurement synchronization skipped because sync client is unavailable",
|
|
||||||
"measurement_id", measurementID,
|
|
||||||
"mode", mode,
|
|
||||||
"has_data", sample != nil,
|
|
||||||
)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err := syncer(ctx, dataSource, mode, sample); err != nil {
|
|
||||||
logManualMeasurementSyncError(ctx, "manual measurement synchronization failed; local update will continue",
|
|
||||||
"measurement_id", measurementID,
|
|
||||||
"mode", mode,
|
|
||||||
"has_data", sample != nil,
|
|
||||||
"error", err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func logManualMeasurementSyncError(ctx context.Context, message string, fields ...any) {
|
|
||||||
// The application initializes logging before serving requests. This guard keeps
|
|
||||||
// the best-effort path safe in isolated unit tests and other pre-init callers.
|
|
||||||
if logger.GetLoggerInstance() == nil {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
logger.Error(ctx, message, fields...)
|
|
||||||
}
|
|
||||||
|
|
||||||
func measurementUpdateNow(dependencies measurementUpdateDependencies) time.Time {
|
|
||||||
if dependencies.nowFunc != nil {
|
|
||||||
return dependencies.nowFunc().UTC()
|
|
||||||
}
|
|
||||||
return time.Now().UTC()
|
|
||||||
}
|
|
||||||
|
|
||||||
func measurementFailureTime(result measurementUpdateResult) time.Time {
|
|
||||||
if !result.operationTime.IsZero() {
|
|
||||||
return result.operationTime
|
|
||||||
}
|
|
||||||
return time.Now().UTC()
|
|
||||||
}
|
|
||||||
|
|
||||||
func parseOptionalMeasurementModeData(raw json.RawMessage) (*float64, error) {
|
func parseOptionalMeasurementModeData(raw json.RawMessage) (*float64, error) {
|
||||||
trimmed := bytes.TrimSpace(raw)
|
trimmed := bytes.TrimSpace(raw)
|
||||||
if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
|
if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
|
||||||
|
|
@ -501,3 +382,44 @@ func isInvalidDataObjectUpdateError(err error) bool {
|
||||||
errors.Is(err, common.ErrMeasurementTokenNotFound) ||
|
errors.Is(err, common.ErrMeasurementTokenNotFound) ||
|
||||||
errors.Is(err, common.ErrAmbiguousMeasurementToken)
|
errors.Is(err, common.ErrAmbiguousMeasurementToken)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func writeMeasurementManualValue(ctx context.Context, measurement *orm.Measurement, value float64) error {
|
||||||
|
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("generate measurement redis key: %w", err)
|
||||||
|
}
|
||||||
|
zset, err := diagram.NewRedisZSet(ctx, key, 0, false)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("create measurement redis zset: %w", err)
|
||||||
|
}
|
||||||
|
if err := zset.ZADD(key, value, strconv.FormatInt(time.Now().UnixNano(), 10)); err != nil {
|
||||||
|
return fmt.Errorf("write manual measurement value to redis: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func callRealTimeDataWriteStopInterface(_ context.Context, _ orm.JSONMap, _ *float64) error {
|
||||||
|
// TODO: call the dataRT HTTP API. A nil value stops automatic writes;
|
||||||
|
// a non-nil value writes the supplied manual measurement value.
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func callRealTimeDataWriteStartInterface(_ context.Context, _ orm.JSONMap, _ *float64) error {
|
||||||
|
// TODO: call the dataRT HTTP API to start automatic measurement writes.
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func replaceMeasurementRedisValue(ctx context.Context, measurement *orm.Measurement, value float64) error {
|
||||||
|
key, err := model.GenerateMeasureIdentifier(measurement.DataSource)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("generate measurement redis key: %w", err)
|
||||||
|
}
|
||||||
|
zset, err := diagram.NewRedisZSet(ctx, key, 0, false)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("create measurement redis zset: %w", err)
|
||||||
|
}
|
||||||
|
if err := zset.ZREPLACE(key, value, strconv.FormatInt(time.Now().UnixNano(), 10)); err != nil {
|
||||||
|
return fmt.Errorf("replace manual measurement value in redis: %w", err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -5,9 +5,7 @@ import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
|
|
||||||
"modelRT/client/manualsync"
|
|
||||||
"modelRT/common/errcode"
|
"modelRT/common/errcode"
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
"modelRT/orm"
|
"modelRT/orm"
|
||||||
|
|
@ -28,7 +26,6 @@ func TestValidateDataObjectAttributeUpdateParameterGroups(t *testing.T) {
|
||||||
"craft",
|
"craft",
|
||||||
"integrity",
|
"integrity",
|
||||||
"behavior",
|
"behavior",
|
||||||
"base_extend",
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, group := range groups {
|
for _, group := range groups {
|
||||||
|
|
@ -49,7 +46,7 @@ func TestValidateDataObjectAttributeUpdateParameterGroups(t *testing.T) {
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestValidateDataObjectAttributeUpdateRejectsUnsupportedParameterGroups(t *testing.T) {
|
func TestValidateDataObjectAttributeUpdateRejectsUnsupportedParameterGroups(t *testing.T) {
|
||||||
for _, group := range []string{"component"} {
|
for _, group := range []string{"component", "base_extend"} {
|
||||||
t.Run(group, func(t *testing.T) {
|
t.Run(group, func(t *testing.T) {
|
||||||
_, _, _, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{
|
_, _, _, err := validateDataObjectAttributeUpdate(dataObjectAttributeUpdateRequest{
|
||||||
Token: fmt.Sprintf("nspath.component.%s.attribute", group),
|
Token: fmt.Sprintf("nspath.component.%s.attribute", group),
|
||||||
|
|
@ -195,11 +192,10 @@ func TestUpdateMeasurementDataObjectLocksRowAndUpdatesMode(t *testing.T) {
|
||||||
|
|
||||||
startCalled := false
|
startCalled := false
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
|
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
|
||||||
syncManualChangeFunc: func(_ context.Context, dataSource orm.JSONMap, mode int16, sample *manualsync.SyntheticData) error {
|
startDataRTFunc: func(_ context.Context, dataSource orm.JSONMap, value *float64) error {
|
||||||
startCalled = true
|
startCalled = true
|
||||||
assert.Equal(t, float64(1), dataSource["type"])
|
assert.Equal(t, float64(1), dataSource["type"])
|
||||||
assert.Equal(t, constants.MeasurementModeAutomatic, mode)
|
assert.Nil(t, value)
|
||||||
assert.Nil(t, sample)
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
@ -210,7 +206,7 @@ func TestUpdateMeasurementDataObjectLocksRowAndUpdatesMode(t *testing.T) {
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
require.NoError(t, mock.ExpectationsWereMet())
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestUpdateMeasurementModeToAutomaticContinuesWhenManualSyncFails(t *testing.T) {
|
func TestUpdateMeasurementModeToAutomaticReturnsErrorWhenDataRTStartFails(t *testing.T) {
|
||||||
db, mock, closeDB := newDataObjectUpdateTestDB(t)
|
db, mock, closeDB := newDataObjectUpdateTestDB(t)
|
||||||
defer closeDB()
|
defer closeDB()
|
||||||
|
|
||||||
|
|
@ -223,14 +219,13 @@ func TestUpdateMeasurementModeToAutomaticContinuesWhenManualSyncFails(t *testing
|
||||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||||
mock.ExpectRollback()
|
mock.ExpectRollback()
|
||||||
|
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
|
_, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeAutomatic, nil, measurementUpdateDependencies{
|
||||||
syncManualChangeFunc: func(context.Context, orm.JSONMap, int16, *manualsync.SyntheticData) error {
|
startDataRTFunc: func(context.Context, orm.JSONMap, *float64) error {
|
||||||
return fmt.Errorf("dataRT unavailable")
|
return fmt.Errorf("dataRT unavailable")
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.Error(t, err)
|
||||||
assert.True(t, result.modeChanged)
|
assert.Contains(t, err.Error(), "start automatic measurement write")
|
||||||
assert.Equal(t, constants.MeasurementModeAutomatic, result.mode)
|
|
||||||
require.NoError(t, tx.Rollback().Error)
|
require.NoError(t, tx.Rollback().Error)
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
require.NoError(t, mock.ExpectationsWereMet())
|
||||||
}
|
}
|
||||||
|
|
@ -268,14 +263,13 @@ func TestUpdateMeasurementModeToManualWithoutDataOnlyStopsDataRT(t *testing.T) {
|
||||||
stopCalled := false
|
stopCalled := false
|
||||||
replaceCalled := false
|
replaceCalled := false
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, nil, measurementUpdateDependencies{
|
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, nil, measurementUpdateDependencies{
|
||||||
syncManualChangeFunc: func(_ context.Context, dataSource orm.JSONMap, mode int16, sample *manualsync.SyntheticData) error {
|
updateDataRTFunc: func(_ context.Context, dataSource orm.JSONMap, value *float64) error {
|
||||||
stopCalled = true
|
stopCalled = true
|
||||||
assert.Equal(t, float64(1), dataSource["type"])
|
assert.Equal(t, float64(1), dataSource["type"])
|
||||||
assert.Equal(t, constants.MeasurementModeManual, mode)
|
assert.Nil(t, value)
|
||||||
assert.Nil(t, sample)
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64, time.Time) error {
|
replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64) error {
|
||||||
replaceCalled = true
|
replaceCalled = true
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
|
|
@ -303,13 +297,12 @@ func TestUpdateMeasurementModeToManualReplacesRedisValueWhenDataProvided(t *test
|
||||||
|
|
||||||
callOrder := make([]string, 0, 2)
|
callOrder := make([]string, 0, 2)
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`0`), measurementUpdateDependencies{
|
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`0`), measurementUpdateDependencies{
|
||||||
syncManualChangeFunc: func(_ context.Context, _ orm.JSONMap, mode int16, sample *manualsync.SyntheticData) error {
|
updateDataRTFunc: func(_ context.Context, _ orm.JSONMap, value *float64) error {
|
||||||
callOrder = append(callOrder, "sync-mode")
|
callOrder = append(callOrder, "stop-dataRT")
|
||||||
assert.Equal(t, constants.MeasurementModeManual, mode)
|
assert.Nil(t, value)
|
||||||
assert.Nil(t, sample)
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
replaceRedisValueFunc: func(_ context.Context, measurement *orm.Measurement, value float64, _ time.Time) error {
|
replaceRedisValueFunc: func(_ context.Context, measurement *orm.Measurement, value float64) error {
|
||||||
callOrder = append(callOrder, "replace-redis")
|
callOrder = append(callOrder, "replace-redis")
|
||||||
assert.Equal(t, int64(10), measurement.ID)
|
assert.Equal(t, int64(10), measurement.ID)
|
||||||
assert.Equal(t, float64(0), value)
|
assert.Equal(t, float64(0), value)
|
||||||
|
|
@ -317,13 +310,13 @@ func TestUpdateMeasurementModeToManualReplacesRedisValueWhenDataProvided(t *test
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, []string{"sync-mode", "replace-redis"}, callOrder)
|
assert.Equal(t, []string{"stop-dataRT", "replace-redis"}, callOrder)
|
||||||
assert.Contains(t, result.message, "manual")
|
assert.Contains(t, result.message, "manual")
|
||||||
require.NoError(t, tx.Rollback().Error)
|
require.NoError(t, tx.Rollback().Error)
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
require.NoError(t, mock.ExpectationsWereMet())
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestUpdateMeasurementModeToManualStillUpdatesRedisWhenManualSyncFails(t *testing.T) {
|
func TestUpdateMeasurementModeToManualDoesNotTouchRedisWhenDataRTStopFails(t *testing.T) {
|
||||||
db, mock, closeDB := newDataObjectUpdateTestDB(t)
|
db, mock, closeDB := newDataObjectUpdateTestDB(t)
|
||||||
defer closeDB()
|
defer closeDB()
|
||||||
|
|
||||||
|
|
@ -337,20 +330,19 @@ func TestUpdateMeasurementModeToManualStillUpdatesRedisWhenManualSyncFails(t *te
|
||||||
mock.ExpectRollback()
|
mock.ExpectRollback()
|
||||||
|
|
||||||
replaceCalled := false
|
replaceCalled := false
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`15.2`), measurementUpdateDependencies{
|
_, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "mode", constants.MeasurementModeManual, json.RawMessage(`15.2`), measurementUpdateDependencies{
|
||||||
syncManualChangeFunc: func(_ context.Context, _ orm.JSONMap, mode int16, sample *manualsync.SyntheticData) error {
|
updateDataRTFunc: func(_ context.Context, _ orm.JSONMap, value *float64) error {
|
||||||
assert.Equal(t, constants.MeasurementModeManual, mode)
|
assert.Nil(t, value)
|
||||||
assert.Nil(t, sample)
|
|
||||||
return fmt.Errorf("dataRT unavailable")
|
return fmt.Errorf("dataRT unavailable")
|
||||||
},
|
},
|
||||||
replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64, time.Time) error {
|
replaceRedisValueFunc: func(context.Context, *orm.Measurement, float64) error {
|
||||||
replaceCalled = true
|
replaceCalled = true
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.Error(t, err)
|
||||||
assert.True(t, replaceCalled)
|
assert.Contains(t, err.Error(), "stop automatic measurement write")
|
||||||
assert.True(t, result.modeChanged)
|
assert.False(t, replaceCalled)
|
||||||
require.NoError(t, tx.Rollback().Error)
|
require.NoError(t, tx.Rollback().Error)
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
require.NoError(t, mock.ExpectationsWereMet())
|
||||||
}
|
}
|
||||||
|
|
@ -402,63 +394,29 @@ func TestUpdateMeasurementDataObjectWritesValueInManualMode(t *testing.T) {
|
||||||
mock.ExpectRollback()
|
mock.ExpectRollback()
|
||||||
|
|
||||||
called := false
|
called := false
|
||||||
operationTime := time.Date(2026, time.August, 4, 10, 0, 0, 123, time.UTC)
|
writer := func(_ context.Context, measurement *orm.Measurement, value float64) error {
|
||||||
writer := func(_ context.Context, measurement *orm.Measurement, value float64, timestamp time.Time) error {
|
|
||||||
called = true
|
called = true
|
||||||
assert.Equal(t, int64(10), measurement.ID)
|
assert.Equal(t, int64(10), measurement.ID)
|
||||||
assert.Equal(t, float64(15.2), value)
|
assert.Equal(t, float64(15.2), value)
|
||||||
assert.Equal(t, operationTime, timestamp)
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
dataRTCalled := false
|
dataRTCalled := false
|
||||||
dataRTWriter := func(_ context.Context, dataSource orm.JSONMap, mode int16, sample *manualsync.SyntheticData) error {
|
dataRTWriter := func(_ context.Context, dataSource orm.JSONMap, value *float64) error {
|
||||||
dataRTCalled = true
|
dataRTCalled = true
|
||||||
assert.Equal(t, constants.MeasurementModeManual, mode)
|
require.NotNil(t, value)
|
||||||
require.NotNil(t, sample)
|
assert.Equal(t, float64(15.2), *value)
|
||||||
assert.Equal(t, float64(15.2), sample.Value)
|
|
||||||
assert.Equal(t, operationTime.UnixNano(), sample.Time)
|
|
||||||
assert.Equal(t, float64(1), dataSource["type"])
|
assert.Equal(t, float64(1), dataSource["type"])
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, measurementUpdateDependencies{
|
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, measurementUpdateDependencies{
|
||||||
writeManualValueFunc: writer,
|
writeManualValueFunc: writer,
|
||||||
syncManualChangeFunc: dataRTWriter,
|
updateDataRTFunc: dataRTWriter,
|
||||||
nowFunc: func() time.Time { return operationTime },
|
|
||||||
})
|
})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.True(t, called)
|
assert.True(t, called)
|
||||||
assert.True(t, dataRTCalled)
|
assert.True(t, dataRTCalled)
|
||||||
assert.Contains(t, result.message, "updated")
|
assert.Contains(t, result.message, "updated")
|
||||||
assert.True(t, result.recordFailureOnError)
|
assert.False(t, result.recordFailure)
|
||||||
assert.Equal(t, int64(10), result.measurementID)
|
|
||||||
assert.Equal(t, float64(15.2), result.value)
|
|
||||||
require.NoError(t, tx.Rollback().Error)
|
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestUpdateMeasurementDataObjectContinuesValueUpdateWhenManualSyncFails(t *testing.T) {
|
|
||||||
db, mock, closeDB := newDataObjectUpdateTestDB(t)
|
|
||||||
defer closeDB()
|
|
||||||
|
|
||||||
mock.ExpectBegin()
|
|
||||||
tx := db.Begin()
|
|
||||||
require.NoError(t, tx.Error)
|
|
||||||
expectMeasurementResolution(mock, constants.MeasurementModeManual)
|
|
||||||
mock.ExpectExec(`UPDATE "measurement" SET "operations"=.*WHERE id = \$3`).
|
|
||||||
WithArgs(sqlmock.AnyArg(), 500, int64(10)).
|
|
||||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
|
||||||
mock.ExpectRollback()
|
|
||||||
|
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, measurementUpdateDependencies{
|
|
||||||
writeManualValueFunc: func(context.Context, *orm.Measurement, float64, time.Time) error {
|
|
||||||
return nil
|
|
||||||
},
|
|
||||||
syncManualChangeFunc: func(context.Context, orm.JSONMap, int16, *manualsync.SyntheticData) error {
|
|
||||||
return fmt.Errorf("manual sync unavailable")
|
|
||||||
},
|
|
||||||
})
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Contains(t, result.message, "updated")
|
|
||||||
require.NoError(t, tx.Rollback().Error)
|
require.NoError(t, tx.Rollback().Error)
|
||||||
require.NoError(t, mock.ExpectationsWereMet())
|
require.NoError(t, mock.ExpectationsWereMet())
|
||||||
}
|
}
|
||||||
|
|
@ -474,14 +432,14 @@ func TestUpdateMeasurementDataObjectReturnsFailureResultAndAppError(t *testing.T
|
||||||
mock.ExpectRollback()
|
mock.ExpectRollback()
|
||||||
|
|
||||||
writeErr := fmt.Errorf("write value failed")
|
writeErr := fmt.Errorf("write value failed")
|
||||||
writer := func(context.Context, *orm.Measurement, float64, time.Time) error { return writeErr }
|
writer := func(context.Context, *orm.Measurement, float64) error { return writeErr }
|
||||||
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, measurementUpdateDependencies{
|
result, err := updateMeasurementDataObject(context.Background(), tx, "nspath.measurement", "value", float64(15.2), nil, measurementUpdateDependencies{
|
||||||
writeManualValueFunc: writer,
|
writeManualValueFunc: writer,
|
||||||
})
|
})
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.ErrorIs(t, err, errcode.ErrMeasurementValueUpdateFailed)
|
assert.ErrorIs(t, err, errcode.ErrMeasurementValueUpdateFailed)
|
||||||
assert.ErrorIs(t, err, writeErr)
|
assert.ErrorIs(t, err, writeErr)
|
||||||
assert.True(t, result.recordFailureOnError)
|
assert.True(t, result.recordFailure)
|
||||||
assert.Equal(t, int64(10), result.measurementID)
|
assert.Equal(t, int64(10), result.measurementID)
|
||||||
assert.Equal(t, float64(15.2), result.value)
|
assert.Equal(t, float64(15.2), result.value)
|
||||||
require.NoError(t, tx.Rollback().Error)
|
require.NoError(t, tx.Rollback().Error)
|
||||||
|
|
|
||||||
73
main.go
73
main.go
|
|
@ -14,7 +14,6 @@ import (
|
||||||
"syscall"
|
"syscall"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"modelRT/client/manualsync"
|
|
||||||
"modelRT/config"
|
"modelRT/config"
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
"modelRT/database"
|
"modelRT/database"
|
||||||
|
|
@ -99,11 +98,10 @@ func main() {
|
||||||
logger.InitLoggerInstance(modelRTConfig.LoggerConfig)
|
logger.InitLoggerInstance(modelRTConfig.LoggerConfig)
|
||||||
defer logger.GetLoggerInstance().Sync()
|
defer logger.GetLoggerInstance().Sync()
|
||||||
|
|
||||||
baseCtx := context.Background()
|
|
||||||
// init OTel TracerProvider
|
// init OTel TracerProvider
|
||||||
tp, tpErr := middleware.InitTracerProvider(context.Background(), modelRTConfig)
|
tp, tpErr := middleware.InitTracerProvider(context.Background(), modelRTConfig)
|
||||||
if tpErr != nil {
|
if tpErr != nil {
|
||||||
logger.Error(baseCtx, "init OTLP tracer provider failed, tracing disabled", "error", tpErr)
|
log.Printf("warn: OTLP tracer init failed, tracing disabled: %v", tpErr)
|
||||||
}
|
}
|
||||||
if tp != nil {
|
if tp != nil {
|
||||||
defer func() {
|
defer func() {
|
||||||
|
|
@ -113,7 +111,7 @@ func main() {
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, startupSpan := otel.Tracer("modelRT/main").Start(baseCtx, "startup")
|
ctx, startupSpan := otel.Tracer("modelRT/main").Start(context.Background(), "startup")
|
||||||
defer startupSpan.End()
|
defer startupSpan.End()
|
||||||
|
|
||||||
hostName, err := os.Hostname()
|
hostName, err := os.Hostname()
|
||||||
|
|
@ -128,16 +126,6 @@ func main() {
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
manualSyncClient, err := manualsync.NewClient(modelRTConfig.DataRTConfig.ManualSync)
|
|
||||||
if err != nil {
|
|
||||||
// TODO: This is a best-effort integration and may not be the final design.
|
|
||||||
// The current behavior lets modelRT start without a manual-sync client;
|
|
||||||
// affected updates still succeed and log an error while sync events are lost.
|
|
||||||
// Revisit fail-fast validation if this downstream service becomes mandatory.
|
|
||||||
logger.Error(ctx, "init manual measurement sync client failed", "error", err)
|
|
||||||
}
|
|
||||||
manualsync.SetDefaultSyncer(manualSyncClient)
|
|
||||||
|
|
||||||
// init postgresDBClient
|
// init postgresDBClient
|
||||||
postgresDBClient = database.InitPostgresDBInstance(ctx, modelRTConfig.PostgresDBURI)
|
postgresDBClient = database.InitPostgresDBInstance(ctx, modelRTConfig.PostgresDBURI)
|
||||||
|
|
||||||
|
|
@ -203,7 +191,7 @@ func main() {
|
||||||
// async push task message to rabbitMQ
|
// async push task message to rabbitMQ
|
||||||
go task.PushTaskToRabbitMQ(ctx, modelRTConfig.RabbitMQConfig, task.TaskMsgChan)
|
go task.PushTaskToRabbitMQ(ctx, modelRTConfig.RabbitMQConfig, task.TaskMsgChan)
|
||||||
|
|
||||||
if err := postgresDBClient.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
postgresDBClient.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||||
// load circuit diagram from postgres
|
// load circuit diagram from postgres
|
||||||
// componentTypeMap, err := database.QueryCircuitDiagramComponentFromDB(cancelCtx, tx, parsePool)
|
// componentTypeMap, err := database.QueryCircuitDiagramComponentFromDB(cancelCtx, tx, parsePool)
|
||||||
// if err != nil {
|
// if err != nil {
|
||||||
|
|
@ -213,80 +201,67 @@ func main() {
|
||||||
|
|
||||||
cacheMap, err := model.GetNSpathToIsLocalMap(ctx, postgresDBClient)
|
cacheMap, err := model.GetNSpathToIsLocalMap(ctx, postgresDBClient)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("get nspath to is_local map: %w", err)
|
logger.Error(ctx, "get nspath to is_local map failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
model.NSPathToIsLocalMap = cacheMap
|
model.NSPathToIsLocalMap = cacheMap
|
||||||
|
|
||||||
err = model.CleanupRecommendRedisCache(ctx)
|
err = model.CleanupRecommendRedisCache(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("clean up component measurement and attribute group: %w", err)
|
logger.Error(ctx, "clean up component measurement and attribute group failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
measurementSet, err := database.GetFullMeasurementSet(ctx, postgresDBClient)
|
measurementSet, err := database.GetFullMeasurementSet(ctx, postgresDBClient)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("generate component measurement group: %w", err)
|
logger.Error(ctx, "generate component measurement group failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet)
|
fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("store component measurement group into redis: %w", err)
|
logger.Error(ctx, "store component measurement group into redis failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
compAttrSet, err := database.GenAllAttributeMap(tx)
|
compAttrSet, err := database.GenAllAttributeMap(tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("generate component attribute group: %w", err)
|
logger.Error(ctx, "generate component attribute group failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet)
|
err = model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("store component attribute group into redis: %w", err)
|
logger.Error(ctx, "store component attribute group into redis failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
componentColumnNames, err := database.QueryComponentColumnNames(ctx, tx)
|
componentColumnNames, err := database.QueryComponentColumnNames(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("query component table column names: %w", err)
|
logger.Error(ctx, "query component table column names failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = model.StoreComponentColumnRecommend(ctx, fullParentPath, isLocalParentPath, componentColumnNames)
|
err = model.StoreComponentColumnRecommend(ctx, fullParentPath, isLocalParentPath, componentColumnNames)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("store component column recommend content: %w", err)
|
logger.Error(ctx, "store component column recommend content failed", "error", err)
|
||||||
}
|
panic(err)
|
||||||
|
|
||||||
parameterRecords, err := database.QueryParameterInitializationRecords(ctx, tx)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("load parameter data objects from postgres: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = model.InitializeParameterDataObjects(ctx, parameterRecords)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("initialize parameter data objects: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
measurementRecords, err := database.QueryMeasurementInitializationRecords(ctx, tx)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("load measurement data objects from postgres: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = model.InitializeMeasurementDataObjects(ctx, measurementRecords)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("initialize measurement data objects: %w", err)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
allMeasurement, err := database.GetAllMeasurements(ctx, tx)
|
allMeasurement, err := database.GetAllMeasurements(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("load measurements from postgres: %w", err)
|
logger.Error(ctx, "load topologic info from postgres failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement)
|
go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement)
|
||||||
|
|
||||||
topologics, err := database.QueryTopologic(ctx, tx)
|
topologics, err := database.QueryTopologic(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("load topologic info from postgres: %w", err)
|
logger.Error(ctx, "load topologic info from postgres failed", "error", err)
|
||||||
|
panic(err)
|
||||||
}
|
}
|
||||||
diagram.SetGlobalTopologyGraph(diagram.NewTopologyGraph(topologics))
|
diagram.SetGlobalTopologyGraph(diagram.NewTopologyGraph(topologics))
|
||||||
return nil
|
return nil
|
||||||
}); err != nil {
|
})
|
||||||
logger.Error(ctx, "initialize modelRT startup data failed", "error", err)
|
|
||||||
panic(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// use release mode in production
|
// use release mode in production
|
||||||
if modelRTConfig.DeployEnv == constants.ProductionDeployMode {
|
if modelRTConfig.DeployEnv == constants.ProductionDeployMode {
|
||||||
|
|
|
||||||
|
|
@ -1,78 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
|
|
||||||
"modelRT/common"
|
|
||||||
"modelRT/constants"
|
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
)
|
|
||||||
|
|
||||||
// DataObjectRedisAliasKey returns the Redis string key used to resolve any
|
|
||||||
// supported token form to the one canonical full-token hash.
|
|
||||||
func DataObjectRedisAliasKey(dataObjectType constants.DataObjectType, token string) (string, error) {
|
|
||||||
switch dataObjectType {
|
|
||||||
case constants.DataObjectTypeParameter:
|
|
||||||
return constants.RedisParameterDataObjectAliasPrefix + token, nil
|
|
||||||
case constants.DataObjectTypeMeasurement:
|
|
||||||
return constants.RedisMeasurementDataObjectAliasPrefix + token, nil
|
|
||||||
default:
|
|
||||||
return "", fmt.Errorf("unsupported data object type %q", dataObjectType)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// ResolveDataObjectRedisKey resolves a full or short token to the canonical
|
|
||||||
// full-token Redis hash created during data-object initialization.
|
|
||||||
func ResolveDataObjectRedisKey(
|
|
||||||
ctx context.Context,
|
|
||||||
rdb *redis.Client,
|
|
||||||
dataObjectType constants.DataObjectType,
|
|
||||||
token string,
|
|
||||||
) (string, error) {
|
|
||||||
if rdb == nil {
|
|
||||||
return "", fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
classifiedType, err := ClassifyDataObjectToken(token)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
if classifiedType != dataObjectType {
|
|
||||||
return "", fmt.Errorf("token %q is %q, expected %q", token, classifiedType, dataObjectType)
|
|
||||||
}
|
|
||||||
|
|
||||||
aliasKey, err := DataObjectRedisAliasKey(dataObjectType, token)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
canonicalKey, err := rdb.Get(ctx, aliasKey).Result()
|
|
||||||
if errors.Is(err, redis.Nil) {
|
|
||||||
switch dataObjectType {
|
|
||||||
case constants.DataObjectTypeParameter:
|
|
||||||
return "", fmt.Errorf("%w: %q", common.ErrParameterTokenNotFound, token)
|
|
||||||
case constants.DataObjectTypeMeasurement:
|
|
||||||
return "", fmt.Errorf("%w: %q", common.ErrMeasurementTokenNotFound, token)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
return "", fmt.Errorf("resolve redis data-object alias %q: %w", token, err)
|
|
||||||
}
|
|
||||||
if canonicalKey == "" {
|
|
||||||
return "", fmt.Errorf("redis data-object alias %q points to an empty key", token)
|
|
||||||
}
|
|
||||||
keyType, err := rdb.Type(ctx, canonicalKey).Result()
|
|
||||||
if err != nil {
|
|
||||||
return "", fmt.Errorf("query canonical redis key type for %q: %w", token, err)
|
|
||||||
}
|
|
||||||
if keyType != "hash" {
|
|
||||||
return "", fmt.Errorf(
|
|
||||||
"redis data-object alias %q points to key %q with type %q, expected hash",
|
|
||||||
token,
|
|
||||||
canonicalKey,
|
|
||||||
keyType,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
return canonicalKey, nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,25 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestDataObjectRedisAliasKey(t *testing.T) {
|
|
||||||
parameterKey, err := DataObjectRedisAliasKey(constants.DataObjectTypeParameter, "nspath.component.rated.voltage")
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, constants.RedisParameterDataObjectAliasPrefix+"nspath.component.rated.voltage", parameterKey)
|
|
||||||
|
|
||||||
measurementKey, err := DataObjectRedisAliasKey(constants.DataObjectTypeMeasurement, "nspath.current")
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, constants.RedisMeasurementDataObjectAliasPrefix+"nspath.current", measurementKey)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestDataObjectRedisAliasKeyRejectsUnsupportedType(t *testing.T) {
|
|
||||||
_, err := DataObjectRedisAliasKey(constants.DataObjectType("unknown"), "token")
|
|
||||||
require.Error(t, err)
|
|
||||||
}
|
|
||||||
|
|
@ -9,6 +9,18 @@ import (
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
var parameterAttributeGroups = map[string]struct{}{
|
||||||
|
"component": {},
|
||||||
|
"base_extend": {},
|
||||||
|
"rated": {},
|
||||||
|
"setup": {},
|
||||||
|
"model": {},
|
||||||
|
"stable": {},
|
||||||
|
"craft": {},
|
||||||
|
"integrity": {},
|
||||||
|
"behavior": {},
|
||||||
|
}
|
||||||
|
|
||||||
// ClassifyDataObjectToken determines whether token identifies a parameter or a
|
// ClassifyDataObjectToken determines whether token identifies a parameter or a
|
||||||
// measurement. Seven-part and four-part tokens are classified by token6, while
|
// measurement. Seven-part and four-part tokens are classified by token6, while
|
||||||
// two-part tokens are treated as measurements at the current stage.
|
// two-part tokens are treated as measurements at the current stage.
|
||||||
|
|
@ -28,7 +40,7 @@ func ClassifyDataObjectToken(token string) (constants.DataObjectType, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
token6 := parts[token6Index]
|
token6 := parts[token6Index]
|
||||||
if constants.IsSupportedParameterAttributeGroup(token6) {
|
if _, ok := parameterAttributeGroups[token6]; ok {
|
||||||
return constants.DataObjectTypeParameter, nil
|
return constants.DataObjectTypeParameter, nil
|
||||||
}
|
}
|
||||||
if token6 == "bay" {
|
if token6 == "bay" {
|
||||||
|
|
|
||||||
|
|
@ -2,27 +2,50 @@ package model
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
|
"modelRT/orm"
|
||||||
)
|
)
|
||||||
|
|
||||||
// MeasurementTypeString converts measurement.type from PostgreSQL to the
|
var allowedMeasurementTypes = map[string]struct{}{
|
||||||
// electric-element type stored in the Redis data-object hash.
|
"TM": {},
|
||||||
func MeasurementTypeString(measurementType int16) (string, error) {
|
"TS": {},
|
||||||
switch measurementType {
|
"TC": {},
|
||||||
case constants.MeasurementTypeTelemetry:
|
"TA": {},
|
||||||
return "TM", nil
|
"SP": {},
|
||||||
case constants.MeasurementTypeTelesignal:
|
}
|
||||||
return "TS", nil
|
|
||||||
case constants.MeasurementTypeTelecommand:
|
// MeasurementTypeFromDataSource returns the two-character measurement type
|
||||||
return "TC", nil
|
// encoded in a CL3611 channel. Only TM, TS, TC, TA, and SP are valid.
|
||||||
case constants.MeasurementTypeTeleadjusting:
|
func MeasurementTypeFromDataSource(dataSource orm.JSONMap) (string, error) {
|
||||||
return "TA", nil
|
dataSourceType, err := integerJSONValue(dataSource["type"])
|
||||||
case constants.MeasurementTypeSetpoint:
|
if err != nil {
|
||||||
return "SP", nil
|
return "", fmt.Errorf("invalid measurement data_source type: %w", err)
|
||||||
default:
|
|
||||||
return "", fmt.Errorf("unsupported measurement type %d", measurementType)
|
|
||||||
}
|
}
|
||||||
|
if dataSourceType != constants.DataSourceTypeCL3611 {
|
||||||
|
return "", fmt.Errorf("measurement type requires data_source type %d, got %d", constants.DataSourceTypeCL3611, dataSourceType)
|
||||||
|
}
|
||||||
|
|
||||||
|
ioAddress, ok := dataSource["io_address"].(map[string]any)
|
||||||
|
if !ok {
|
||||||
|
if value, jsonMapOK := dataSource["io_address"].(orm.JSONMap); jsonMapOK {
|
||||||
|
ioAddress = map[string]any(value)
|
||||||
|
} else {
|
||||||
|
return "", fmt.Errorf("measurement data_source io_address is not an object")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
channel, ok := ioAddress["channel"].(string)
|
||||||
|
if !ok || len(channel) < 2 {
|
||||||
|
return "", fmt.Errorf("measurement data_source channel must contain at least two characters")
|
||||||
|
}
|
||||||
|
|
||||||
|
measurementType := strings.ToUpper(channel[:2])
|
||||||
|
if _, ok := allowedMeasurementTypes[measurementType]; !ok {
|
||||||
|
return "", fmt.Errorf("unsupported measurement type %q", measurementType)
|
||||||
|
}
|
||||||
|
return measurementType, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func integerJSONValue(value any) (int, error) {
|
func integerJSONValue(value any) (int, error) {
|
||||||
|
|
|
||||||
|
|
@ -1,37 +1,40 @@
|
||||||
package model
|
package model
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"modelRT/orm"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestMeasurementTypeString(t *testing.T) {
|
func TestMeasurementTypeFromDataSource(t *testing.T) {
|
||||||
tests := []struct {
|
for _, measurementType := range []string{"TM", "TS", "TC", "TA", "SP"} {
|
||||||
value int16
|
t.Run(measurementType, func(t *testing.T) {
|
||||||
expected string
|
actual, err := MeasurementTypeFromDataSource(orm.JSONMap{
|
||||||
}{
|
"type": float64(1),
|
||||||
{value: 0, expected: "TM"},
|
"io_address": map[string]any{
|
||||||
{value: 1, expected: "TS"},
|
"channel": strings.ToLower(measurementType) + "1_test",
|
||||||
{value: 2, expected: "TC"},
|
},
|
||||||
{value: 3, expected: "TA"},
|
})
|
||||||
{value: 4, expected: "SP"},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.expected, func(t *testing.T) {
|
|
||||||
actual, err := MeasurementTypeString(test.value)
|
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.Equal(t, test.expected, actual)
|
assert.Equal(t, measurementType, actual)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMeasurementTypeStringRejectsInvalidValues(t *testing.T) {
|
func TestMeasurementTypeFromDataSourceRejectsInvalidValues(t *testing.T) {
|
||||||
for _, value := range []int16{-1, 5, 100} {
|
tests := []orm.JSONMap{
|
||||||
_, err := MeasurementTypeString(value)
|
{"type": float64(2), "io_address": map[string]any{"channel": "tm1"}},
|
||||||
|
{"type": float64(1), "io_address": map[string]any{"channel": "xx1"}},
|
||||||
|
{"type": float64(1), "io_address": map[string]any{"channel": "t"}},
|
||||||
|
{"type": "1", "io_address": map[string]any{"channel": "tm1"}},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, dataSource := range tests {
|
||||||
|
_, err := MeasurementTypeFromDataSource(dataSource)
|
||||||
require.Error(t, err)
|
require.Error(t, err)
|
||||||
assert.Contains(t, err.Error(), "unsupported measurement type")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,268 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"strings"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/diagram"
|
|
||||||
"modelRT/logger"
|
|
||||||
"modelRT/orm"
|
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
)
|
|
||||||
|
|
||||||
const measurementDataObjectPipelineSize = 500
|
|
||||||
|
|
||||||
type measurementDataObjectHash struct {
|
|
||||||
Key string
|
|
||||||
Aliases []string
|
|
||||||
Fields map[string]any
|
|
||||||
}
|
|
||||||
|
|
||||||
// MeasurementInitializationRecord contains a measurement and the hierarchy
|
|
||||||
// needed to create all supported Redis data-object token aliases.
|
|
||||||
type MeasurementInitializationRecord struct {
|
|
||||||
GridTag string `gorm:"column:grid_tag"`
|
|
||||||
ZoneTag string `gorm:"column:zone_tag"`
|
|
||||||
StationTag string `gorm:"column:station_tag"`
|
|
||||||
ComponentUUID string `gorm:"column:component_uuid"`
|
|
||||||
ComponentNSPath string `gorm:"column:component_nspath"`
|
|
||||||
ComponentTag string `gorm:"column:component_tag"`
|
|
||||||
MeasurementID int64 `gorm:"column:measurement_id"`
|
|
||||||
MeasurementTag string `gorm:"column:measurement_tag"`
|
|
||||||
MeasurementName string `gorm:"column:measurement_name"`
|
|
||||||
MeasurementType int16 `gorm:"column:measurement_type"`
|
|
||||||
MeasurementMode int16 `gorm:"column:measurement_mode"`
|
|
||||||
MeasurementSize int `gorm:"column:measurement_size"`
|
|
||||||
MeasurementDataSource orm.JSONMap `gorm:"column:measurement_data_source;type:jsonb"`
|
|
||||||
MeasurementEventPlan orm.JSONMap `gorm:"column:measurement_event_plan;type:jsonb"`
|
|
||||||
MeasurementBinding orm.JSONMap `gorm:"column:measurement_binding;type:jsonb"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// InitializeMeasurementDataObjects creates one seven-part Redis hash and
|
|
||||||
// aliases for all supported measurement token forms.
|
|
||||||
func InitializeMeasurementDataObjects(ctx context.Context, records []MeasurementInitializationRecord) error {
|
|
||||||
hashes, err := buildMeasurementDataObjectHashes(records)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("build measurement data-object hashes: %w", err)
|
|
||||||
}
|
|
||||||
if err := storeMeasurementDataObjectHashes(ctx, diagram.GetRedisClientInstance(), hashes); err != nil {
|
|
||||||
return fmt.Errorf("store measurement data-object hashes in redis: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.Info(ctx, "initialize measurement data objects completed",
|
|
||||||
"postgres_record_count", len(records),
|
|
||||||
"redis_hash_count", len(hashes),
|
|
||||||
)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func buildMeasurementDataObjectHashes(records []MeasurementInitializationRecord) ([]measurementDataObjectHash, error) {
|
|
||||||
hashes := make([]measurementDataObjectHash, 0, len(records))
|
|
||||||
seenKeys := make(map[string]string, len(records)*3)
|
|
||||||
for _, record := range records {
|
|
||||||
if record.MeasurementMode != constants.MeasurementModeManual &&
|
|
||||||
record.MeasurementMode != constants.MeasurementModeAutomatic {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"measurement %q mode must be %d or %d, got %d",
|
|
||||||
record.MeasurementTag,
|
|
||||||
constants.MeasurementModeManual,
|
|
||||||
constants.MeasurementModeAutomatic,
|
|
||||||
record.MeasurementMode,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if record.MeasurementSize <= 0 {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"measurement %q window size must be greater than 0, got %d",
|
|
||||||
record.MeasurementTag,
|
|
||||||
record.MeasurementSize,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
if record.MeasurementDataSource == nil ||
|
|
||||||
record.MeasurementEventPlan == nil ||
|
|
||||||
record.MeasurementBinding == nil {
|
|
||||||
return nil, fmt.Errorf("measurement %q contains a null JSONB field", record.MeasurementTag)
|
|
||||||
}
|
|
||||||
|
|
||||||
fullToken := strings.Join([]string{
|
|
||||||
record.GridTag,
|
|
||||||
record.ZoneTag,
|
|
||||||
record.StationTag,
|
|
||||||
record.ComponentNSPath,
|
|
||||||
record.ComponentTag,
|
|
||||||
"bay",
|
|
||||||
record.MeasurementTag,
|
|
||||||
}, ".")
|
|
||||||
fourPartToken := strings.Join([]string{
|
|
||||||
record.ComponentNSPath,
|
|
||||||
record.ComponentTag,
|
|
||||||
"bay",
|
|
||||||
record.MeasurementTag,
|
|
||||||
}, ".")
|
|
||||||
twoPartToken := record.ComponentNSPath + "." + record.MeasurementTag
|
|
||||||
|
|
||||||
aliases := []string{fullToken, fourPartToken, twoPartToken}
|
|
||||||
for _, token := range aliases {
|
|
||||||
if err := validateInitializedMeasurementToken(token); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
measurementType, err := MeasurementTypeString(record.MeasurementType)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("derive type for measurement %q: %w", fullToken, err)
|
|
||||||
}
|
|
||||||
dataSource, err := measurementInitializationJSON(record.MeasurementDataSource)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("encode data_source for measurement %q: %w", fullToken, err)
|
|
||||||
}
|
|
||||||
eventPlan, err := measurementInitializationJSON(record.MeasurementEventPlan)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("encode event_plan for measurement %q: %w", fullToken, err)
|
|
||||||
}
|
|
||||||
binding, err := measurementInitializationJSON(record.MeasurementBinding)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("encode binding for measurement %q: %w", fullToken, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
fields := map[string]any{
|
|
||||||
"mode": record.MeasurementMode,
|
|
||||||
"meta": "MEASUREMENT",
|
|
||||||
"type": measurementType,
|
|
||||||
"name": twoPartToken,
|
|
||||||
"description": record.MeasurementName,
|
|
||||||
"id": fullToken,
|
|
||||||
"size": record.MeasurementSize,
|
|
||||||
"data_source": dataSource,
|
|
||||||
"event_plan": eventPlan,
|
|
||||||
"binding": binding,
|
|
||||||
}
|
|
||||||
owner := fmt.Sprintf("%d/%s", record.MeasurementID, record.ComponentUUID)
|
|
||||||
for _, alias := range aliases {
|
|
||||||
if existingOwner, exists := seenKeys[alias]; exists {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"ambiguous measurement token %q is produced by %q and %q",
|
|
||||||
alias,
|
|
||||||
existingOwner,
|
|
||||||
owner,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
seenKeys[alias] = owner
|
|
||||||
}
|
|
||||||
hashes = append(hashes, measurementDataObjectHash{Key: fullToken, Aliases: aliases, Fields: fields})
|
|
||||||
}
|
|
||||||
return hashes, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func validateInitializedMeasurementToken(token string) error {
|
|
||||||
dataObjectType, err := ClassifyDataObjectToken(token)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("generated invalid measurement token %q: %w", token, err)
|
|
||||||
}
|
|
||||||
if dataObjectType != constants.DataObjectTypeMeasurement {
|
|
||||||
return fmt.Errorf("generated token %q is not a measurement", token)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func measurementInitializationJSON(value orm.JSONMap) (string, error) {
|
|
||||||
encoded, err := json.Marshal(value)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
return string(encoded), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func storeMeasurementDataObjectHashes(
|
|
||||||
ctx context.Context,
|
|
||||||
rdb *redis.Client,
|
|
||||||
hashes []measurementDataObjectHash,
|
|
||||||
) error {
|
|
||||||
if rdb == nil {
|
|
||||||
return fmt.Errorf("redis client is nil")
|
|
||||||
}
|
|
||||||
|
|
||||||
oldKeys, err := rdb.SMembers(ctx, constants.RedisMeasurementDataObjectKeySet).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query previously initialized measurement keys: %w", err)
|
|
||||||
}
|
|
||||||
oldAliasKeys, err := rdb.SMembers(ctx, constants.RedisMeasurementDataObjectAliasKeySet).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query previously initialized measurement alias keys: %w", err)
|
|
||||||
}
|
|
||||||
currentKeys := make(map[string]struct{}, len(hashes))
|
|
||||||
currentAliasKeys := make(map[string]struct{}, len(hashes)*3)
|
|
||||||
for start := 0; start < len(hashes); start += measurementDataObjectPipelineSize {
|
|
||||||
end := min(start+measurementDataObjectPipelineSize, len(hashes))
|
|
||||||
pipeline := rdb.TxPipeline()
|
|
||||||
keyMembers := make([]any, 0, end-start)
|
|
||||||
aliasKeyMembers := make([]any, 0, (end-start)*3)
|
|
||||||
for _, hash := range hashes[start:end] {
|
|
||||||
pipeline.Del(ctx, hash.Key)
|
|
||||||
pipeline.HSet(ctx, hash.Key, hash.Fields)
|
|
||||||
keyMembers = append(keyMembers, hash.Key)
|
|
||||||
currentKeys[hash.Key] = struct{}{}
|
|
||||||
for _, alias := range hash.Aliases {
|
|
||||||
aliasKey, err := DataObjectRedisAliasKey(constants.DataObjectTypeMeasurement, alias)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
pipeline.Set(ctx, aliasKey, hash.Key, 0)
|
|
||||||
aliasKeyMembers = append(aliasKeyMembers, aliasKey)
|
|
||||||
currentAliasKeys[aliasKey] = struct{}{}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if len(keyMembers) > 0 {
|
|
||||||
pipeline.SAdd(ctx, constants.RedisMeasurementDataObjectKeySet, keyMembers...)
|
|
||||||
}
|
|
||||||
if len(aliasKeyMembers) > 0 {
|
|
||||||
pipeline.SAdd(ctx, constants.RedisMeasurementDataObjectAliasKeySet, aliasKeyMembers...)
|
|
||||||
}
|
|
||||||
if _, err := pipeline.Exec(ctx); err != nil {
|
|
||||||
return fmt.Errorf("write measurement data-object hash batch starting at %d: %w", start, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
staleKeys := make([]string, 0)
|
|
||||||
for _, key := range oldKeys {
|
|
||||||
if _, exists := currentKeys[key]; !exists {
|
|
||||||
staleKeys = append(staleKeys, key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
cleanupPipeline := rdb.TxPipeline()
|
|
||||||
for start := 0; start < len(staleKeys); start += measurementDataObjectPipelineSize {
|
|
||||||
end := min(start+measurementDataObjectPipelineSize, len(staleKeys))
|
|
||||||
cleanupPipeline.Del(ctx, staleKeys[start:end]...)
|
|
||||||
members := make([]any, 0, end-start)
|
|
||||||
for _, key := range staleKeys[start:end] {
|
|
||||||
members = append(members, key)
|
|
||||||
}
|
|
||||||
cleanupPipeline.SRem(ctx, constants.RedisMeasurementDataObjectKeySet, members...)
|
|
||||||
}
|
|
||||||
staleAliasKeys := make([]string, 0)
|
|
||||||
for _, aliasKey := range oldAliasKeys {
|
|
||||||
if _, exists := currentAliasKeys[aliasKey]; !exists {
|
|
||||||
staleAliasKeys = append(staleAliasKeys, aliasKey)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for start := 0; start < len(staleAliasKeys); start += measurementDataObjectPipelineSize {
|
|
||||||
end := min(start+measurementDataObjectPipelineSize, len(staleAliasKeys))
|
|
||||||
cleanupPipeline.Del(ctx, staleAliasKeys[start:end]...)
|
|
||||||
members := make([]any, 0, end-start)
|
|
||||||
for _, aliasKey := range staleAliasKeys[start:end] {
|
|
||||||
members = append(members, aliasKey)
|
|
||||||
}
|
|
||||||
cleanupPipeline.SRem(ctx, constants.RedisMeasurementDataObjectAliasKeySet, members...)
|
|
||||||
}
|
|
||||||
if len(hashes) == 0 {
|
|
||||||
cleanupPipeline.Del(ctx, constants.RedisMeasurementDataObjectKeySet)
|
|
||||||
cleanupPipeline.Del(ctx, constants.RedisMeasurementDataObjectAliasKeySet)
|
|
||||||
}
|
|
||||||
if _, err := cleanupPipeline.Exec(ctx); err != nil {
|
|
||||||
return fmt.Errorf("remove stale measurement data-object hashes: %w", err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,107 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"modelRT/orm"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestBuildMeasurementDataObjectHashesCreatesCanonicalHashAndAllAliases(t *testing.T) {
|
|
||||||
record := measurementInitializationRecordForTest()
|
|
||||||
|
|
||||||
hashes, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{record})
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Len(t, hashes, 1)
|
|
||||||
|
|
||||||
fullToken := "grid000.zone000.station000.220kV_xuefulu1.CTA.bay.IA_rms"
|
|
||||||
fourPartToken := "220kV_xuefulu1.CTA.bay.IA_rms"
|
|
||||||
twoPartToken := "220kV_xuefulu1.IA_rms"
|
|
||||||
assert.Equal(t, fullToken, hashes[0].Key)
|
|
||||||
assert.Equal(t, []string{fullToken, fourPartToken, twoPartToken}, hashes[0].Aliases)
|
|
||||||
|
|
||||||
fields := hashes[0].Fields
|
|
||||||
assert.NotContains(t, fields, "value")
|
|
||||||
assert.Equal(t, int16(1), fields["mode"])
|
|
||||||
assert.Equal(t, "MEASUREMENT", fields["meta"])
|
|
||||||
assert.Equal(t, "TM", fields["type"])
|
|
||||||
assert.Equal(t, twoPartToken, fields["name"])
|
|
||||||
assert.Equal(t, "A相保护电流有效值", fields["description"])
|
|
||||||
assert.Equal(t, fullToken, fields["id"])
|
|
||||||
assert.Equal(t, 1, fields["size"])
|
|
||||||
assert.Equal(t, `{"io_address":{"channel":"TM1","device":"CTA","dtype":1,"option":"rms","station":"001"},"type":1}`, fields["data_source"])
|
|
||||||
assert.Equal(t, `{}`, fields["event_plan"])
|
|
||||||
assert.Equal(t, `{"ct":{"index":0,"polarity":1,"ratio":1250}}`, fields["binding"])
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBuildMeasurementDataObjectHashesRejectsAmbiguousShortToken(t *testing.T) {
|
|
||||||
first := measurementInitializationRecordForTest()
|
|
||||||
second := first
|
|
||||||
second.GridTag = "grid001"
|
|
||||||
second.ZoneTag = "zone001"
|
|
||||||
second.StationTag = "station001"
|
|
||||||
second.ComponentUUID = "component-uuid-2"
|
|
||||||
second.ComponentTag = "CTB"
|
|
||||||
second.MeasurementID = 2
|
|
||||||
|
|
||||||
_, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{first, second})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "ambiguous measurement token")
|
|
||||||
assert.Contains(t, err.Error(), "220kV_xuefulu1.IA_rms")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBuildMeasurementDataObjectHashesRejectsUnsupportedMeasurementType(t *testing.T) {
|
|
||||||
record := measurementInitializationRecordForTest()
|
|
||||||
record.MeasurementType = 5
|
|
||||||
|
|
||||||
_, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{record})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "derive type")
|
|
||||||
assert.Contains(t, err.Error(), "unsupported measurement type 5")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBuildMeasurementDataObjectHashesRejectsInvalidWindowSize(t *testing.T) {
|
|
||||||
record := measurementInitializationRecordForTest()
|
|
||||||
record.MeasurementSize = 0
|
|
||||||
|
|
||||||
_, err := buildMeasurementDataObjectHashes([]MeasurementInitializationRecord{record})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "window size must be greater than 0")
|
|
||||||
}
|
|
||||||
|
|
||||||
func measurementInitializationRecordForTest() MeasurementInitializationRecord {
|
|
||||||
return MeasurementInitializationRecord{
|
|
||||||
GridTag: "grid000",
|
|
||||||
ZoneTag: "zone000",
|
|
||||||
StationTag: "station000",
|
|
||||||
ComponentUUID: "component-uuid-1",
|
|
||||||
ComponentNSPath: "220kV_xuefulu1",
|
|
||||||
ComponentTag: "CTA",
|
|
||||||
MeasurementID: 1,
|
|
||||||
MeasurementTag: "IA_rms",
|
|
||||||
MeasurementName: "A相保护电流有效值",
|
|
||||||
MeasurementType: 0,
|
|
||||||
MeasurementMode: 1,
|
|
||||||
MeasurementSize: 1,
|
|
||||||
MeasurementDataSource: orm.JSONMap{
|
|
||||||
"type": float64(1),
|
|
||||||
"io_address": map[string]any{
|
|
||||||
"station": "001",
|
|
||||||
"device": "CTA",
|
|
||||||
"channel": "TM1",
|
|
||||||
"dtype": float64(1),
|
|
||||||
"option": "rms",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
MeasurementEventPlan: orm.JSONMap{},
|
|
||||||
MeasurementBinding: orm.JSONMap{
|
|
||||||
"ct": map[string]any{
|
|
||||||
"index": float64(0),
|
|
||||||
"ratio": float64(1250),
|
|
||||||
"polarity": float64(1),
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -11,14 +11,6 @@ import (
|
||||||
"modelRT/constants"
|
"modelRT/constants"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
|
||||||
// CL3611DataSourceTypePhasor define identifies CL3611 phasor data source.
|
|
||||||
CL3611DataSourceTypePhasor = 1
|
|
||||||
|
|
||||||
// CL3611DataSourceTypeSample define identifies CL3611 sampled data source.
|
|
||||||
CL3611DataSourceTypeSample = 2
|
|
||||||
)
|
|
||||||
|
|
||||||
// MeasurementDataSource define measurement data source struct
|
// MeasurementDataSource define measurement data source struct
|
||||||
type MeasurementDataSource struct {
|
type MeasurementDataSource struct {
|
||||||
Type int `json:"type"`
|
Type int `json:"type"`
|
||||||
|
|
@ -230,11 +222,6 @@ func GenerateMeasureIdentifier(source map[string]any) (string, error) {
|
||||||
|
|
||||||
switch regType {
|
switch regType {
|
||||||
case constants.DataSourceTypeCL3611:
|
case constants.DataSourceTypeCL3611:
|
||||||
rawDtype, ok := ioAddress["dtype"].(float64)
|
|
||||||
if !ok {
|
|
||||||
return "", fmt.Errorf("CL3611:invalid or missing dtype field")
|
|
||||||
}
|
|
||||||
|
|
||||||
station, ok := ioAddress["station"].(string)
|
station, ok := ioAddress["station"].(string)
|
||||||
if !ok {
|
if !ok {
|
||||||
return "", fmt.Errorf("CL3611:invalid or missing station field")
|
return "", fmt.Errorf("CL3611:invalid or missing station field")
|
||||||
|
|
@ -248,21 +235,7 @@ func GenerateMeasureIdentifier(source map[string]any) (string, error) {
|
||||||
if !ok {
|
if !ok {
|
||||||
return "", fmt.Errorf("CL3611:invalid or missing channel field")
|
return "", fmt.Errorf("CL3611:invalid or missing channel field")
|
||||||
}
|
}
|
||||||
|
return concatCL361WithPlus(station, device, channel), nil
|
||||||
optinon, ok := ioAddress["option"].(string)
|
|
||||||
if !ok {
|
|
||||||
return "", fmt.Errorf("CL3611:invalid or missing optinon field")
|
|
||||||
}
|
|
||||||
dtype := int(rawDtype)
|
|
||||||
switch dtype {
|
|
||||||
case CL3611DataSourceTypePhasor:
|
|
||||||
return buildCL3611PhasorIdentifier(station, device, channel, optinon), nil
|
|
||||||
case CL3611DataSourceTypeSample:
|
|
||||||
return buildCL3611SampleIdentifier(station, device, channel), nil
|
|
||||||
default:
|
|
||||||
return "", fmt.Errorf("CL3611:unsupported dtype %d", dtype)
|
|
||||||
}
|
|
||||||
|
|
||||||
case constants.DataSourceTypePower104:
|
case constants.DataSourceTypePower104:
|
||||||
station, ok := ioAddress["station"].(string)
|
station, ok := ioAddress["station"].(string)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
|
@ -297,10 +270,6 @@ func concatP104WithPlus(station string, packet int, offset int) string {
|
||||||
return strings.ToLower(station + ":104:" + packetStr + ":" + offsetStr)
|
return strings.ToLower(station + ":104:" + packetStr + ":" + offsetStr)
|
||||||
}
|
}
|
||||||
|
|
||||||
func buildCL3611SampleIdentifier(station, device, channel string) string {
|
func concatCL361WithPlus(station, device, channel string) string {
|
||||||
return strings.ToLower(station + ":" + device + ":" + "phasor" + ":" + channel)
|
return strings.ToLower(station + ":" + device + ":" + "phasor" + ":" + channel)
|
||||||
}
|
}
|
||||||
|
|
||||||
func buildCL3611PhasorIdentifier(station, device, channel, option string) string {
|
|
||||||
return strings.ToLower(station + ":" + device + ":" + "phasor" + ":" + channel + ":" + option)
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -1,263 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"context"
|
|
||||||
"database/sql"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"strings"
|
|
||||||
|
|
||||||
"modelRT/constants"
|
|
||||||
"modelRT/diagram"
|
|
||||||
"modelRT/logger"
|
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
)
|
|
||||||
|
|
||||||
const parameterDataObjectPipelineSize = 500
|
|
||||||
|
|
||||||
type parameterDataObjectHash struct {
|
|
||||||
Key string
|
|
||||||
Aliases []string
|
|
||||||
Fields map[string]any
|
|
||||||
}
|
|
||||||
|
|
||||||
// ParameterInitializationRecord contains one parameter attribute together with
|
|
||||||
// the hierarchy and metadata needed to create its Redis data-object hashes.
|
|
||||||
type ParameterInitializationRecord struct {
|
|
||||||
GridTag string `gorm:"column:grid_tag"`
|
|
||||||
ZoneTag string `gorm:"column:zone_tag"`
|
|
||||||
StationTag string `gorm:"column:station_tag"`
|
|
||||||
StationIsLocal bool `gorm:"column:station_is_local"`
|
|
||||||
ComponentUUID string `gorm:"column:component_uuid"`
|
|
||||||
ComponentNSPath string `gorm:"column:component_nspath"`
|
|
||||||
ComponentTag string `gorm:"column:component_tag"`
|
|
||||||
AttributeGroup string `gorm:"column:attribute_group"`
|
|
||||||
AttributeName string `gorm:"column:attribute_name"`
|
|
||||||
AttributeValue string `gorm:"column:attribute_value"`
|
|
||||||
AttributeType string `gorm:"column:attribute_type"`
|
|
||||||
Description sql.NullString `gorm:"column:description"`
|
|
||||||
DescriptionCount int64 `gorm:"column:description_count"`
|
|
||||||
DynamicRecordCount int64 `gorm:"column:dynamic_record_count"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// InitializeParameterDataObjects creates one full-token Redis hash and full or
|
|
||||||
// local-short token aliases for each parameter loaded from PostgreSQL.
|
|
||||||
func InitializeParameterDataObjects(ctx context.Context, records []ParameterInitializationRecord) error {
|
|
||||||
hashes, err := buildParameterDataObjectHashes(records)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("build parameter data-object hashes: %w", err)
|
|
||||||
}
|
|
||||||
if err := storeParameterDataObjectHashes(ctx, diagram.GetRedisClientInstance(), hashes); err != nil {
|
|
||||||
return fmt.Errorf("store parameter data-object hashes in redis: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
logger.Info(ctx, "initialize parameter data objects completed",
|
|
||||||
"postgres_record_count", len(records),
|
|
||||||
"redis_hash_count", len(hashes),
|
|
||||||
)
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func buildParameterDataObjectHashes(records []ParameterInitializationRecord) ([]parameterDataObjectHash, error) {
|
|
||||||
hashes := make([]parameterDataObjectHash, 0, len(records))
|
|
||||||
seenKeys := make(map[string]string, len(records)*2)
|
|
||||||
|
|
||||||
for _, record := range records {
|
|
||||||
value, err := parameterRedisValue(record.AttributeValue)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"decode value for component %q group %q attribute %q: %w",
|
|
||||||
record.ComponentTag,
|
|
||||||
record.AttributeGroup,
|
|
||||||
record.AttributeName,
|
|
||||||
err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
fullToken := strings.Join([]string{
|
|
||||||
record.GridTag,
|
|
||||||
record.ZoneTag,
|
|
||||||
record.StationTag,
|
|
||||||
record.ComponentNSPath,
|
|
||||||
record.ComponentTag,
|
|
||||||
record.AttributeGroup,
|
|
||||||
record.AttributeName,
|
|
||||||
}, ".")
|
|
||||||
shortToken := strings.Join([]string{
|
|
||||||
record.ComponentNSPath,
|
|
||||||
record.ComponentTag,
|
|
||||||
record.AttributeGroup,
|
|
||||||
record.AttributeName,
|
|
||||||
}, ".")
|
|
||||||
|
|
||||||
aliases := []string{fullToken}
|
|
||||||
if record.StationIsLocal {
|
|
||||||
aliases = append(aliases, shortToken)
|
|
||||||
}
|
|
||||||
fields := map[string]any{
|
|
||||||
"value": value,
|
|
||||||
"meta": "PARAM",
|
|
||||||
"type": record.AttributeType,
|
|
||||||
"name": shortToken,
|
|
||||||
"description": record.Description.String,
|
|
||||||
"id": fullToken,
|
|
||||||
}
|
|
||||||
owner := strings.Join([]string{
|
|
||||||
record.ComponentUUID,
|
|
||||||
record.AttributeGroup,
|
|
||||||
record.AttributeName,
|
|
||||||
}, "/")
|
|
||||||
for _, alias := range aliases {
|
|
||||||
if err := validateInitializedParameterToken(alias); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
if existingOwner, exists := seenKeys[alias]; exists {
|
|
||||||
return nil, fmt.Errorf(
|
|
||||||
"ambiguous parameter token %q is produced by %q and %q",
|
|
||||||
alias,
|
|
||||||
existingOwner,
|
|
||||||
owner,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
seenKeys[alias] = owner
|
|
||||||
}
|
|
||||||
hashes = append(hashes, parameterDataObjectHash{Key: fullToken, Aliases: aliases, Fields: fields})
|
|
||||||
}
|
|
||||||
return hashes, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func validateInitializedParameterToken(token string) error {
|
|
||||||
dataObjectType, err := ClassifyDataObjectToken(token)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("generated invalid parameter token %q: %w", token, err)
|
|
||||||
}
|
|
||||||
if dataObjectType != constants.DataObjectTypeParameter {
|
|
||||||
return fmt.Errorf("generated token %q is not a parameter", token)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func parameterRedisValue(rawJSON string) (any, error) {
|
|
||||||
decoder := json.NewDecoder(bytes.NewBufferString(rawJSON))
|
|
||||||
decoder.UseNumber()
|
|
||||||
|
|
||||||
var value any
|
|
||||||
if err := decoder.Decode(&value); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
switch typedValue := value.(type) {
|
|
||||||
case nil:
|
|
||||||
return "null", nil
|
|
||||||
case string:
|
|
||||||
return typedValue, nil
|
|
||||||
case json.Number:
|
|
||||||
return typedValue.String(), nil
|
|
||||||
case bool:
|
|
||||||
return typedValue, nil
|
|
||||||
default:
|
|
||||||
encoded, err := json.Marshal(typedValue)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
return string(encoded), nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func storeParameterDataObjectHashes(
|
|
||||||
ctx context.Context,
|
|
||||||
rdb *redis.Client,
|
|
||||||
hashes []parameterDataObjectHash,
|
|
||||||
) error {
|
|
||||||
if rdb == nil {
|
|
||||||
return fmt.Errorf("redis client is nil")
|
|
||||||
}
|
|
||||||
|
|
||||||
oldKeys, err := rdb.SMembers(ctx, constants.RedisParameterDataObjectKeySet).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query previously initialized parameter keys: %w", err)
|
|
||||||
}
|
|
||||||
oldAliasKeys, err := rdb.SMembers(ctx, constants.RedisParameterDataObjectAliasKeySet).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query previously initialized parameter alias keys: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
currentKeys := make(map[string]struct{}, len(hashes))
|
|
||||||
currentAliasKeys := make(map[string]struct{}, len(hashes)*2)
|
|
||||||
for start := 0; start < len(hashes); start += parameterDataObjectPipelineSize {
|
|
||||||
end := min(start+parameterDataObjectPipelineSize, len(hashes))
|
|
||||||
pipeline := rdb.TxPipeline()
|
|
||||||
keyMembers := make([]any, 0, end-start)
|
|
||||||
aliasKeyMembers := make([]any, 0, (end-start)*2)
|
|
||||||
for _, hash := range hashes[start:end] {
|
|
||||||
pipeline.Del(ctx, hash.Key)
|
|
||||||
pipeline.HSet(ctx, hash.Key, hash.Fields)
|
|
||||||
keyMembers = append(keyMembers, hash.Key)
|
|
||||||
currentKeys[hash.Key] = struct{}{}
|
|
||||||
for _, alias := range hash.Aliases {
|
|
||||||
aliasKey, err := DataObjectRedisAliasKey(constants.DataObjectTypeParameter, alias)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
pipeline.Set(ctx, aliasKey, hash.Key, 0)
|
|
||||||
aliasKeyMembers = append(aliasKeyMembers, aliasKey)
|
|
||||||
currentAliasKeys[aliasKey] = struct{}{}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if len(keyMembers) > 0 {
|
|
||||||
pipeline.SAdd(ctx, constants.RedisParameterDataObjectKeySet, keyMembers...)
|
|
||||||
}
|
|
||||||
if len(aliasKeyMembers) > 0 {
|
|
||||||
pipeline.SAdd(ctx, constants.RedisParameterDataObjectAliasKeySet, aliasKeyMembers...)
|
|
||||||
}
|
|
||||||
if _, err := pipeline.Exec(ctx); err != nil {
|
|
||||||
return fmt.Errorf("write parameter data-object hash batch starting at %d: %w", start, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := cleanupStaleParameterDataObjectKeys(
|
|
||||||
ctx,
|
|
||||||
rdb,
|
|
||||||
oldKeys,
|
|
||||||
oldAliasKeys,
|
|
||||||
currentKeys,
|
|
||||||
currentAliasKeys,
|
|
||||||
); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func cleanupStaleParameterDataObjectKeys(
|
|
||||||
ctx context.Context,
|
|
||||||
rdb *redis.Client,
|
|
||||||
oldKeys, oldAliasKeys []string,
|
|
||||||
currentKeys, currentAliasKeys map[string]struct{},
|
|
||||||
) error {
|
|
||||||
pipeline := rdb.TxPipeline()
|
|
||||||
for _, key := range oldKeys {
|
|
||||||
if _, exists := currentKeys[key]; exists {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
pipeline.Del(ctx, key)
|
|
||||||
pipeline.SRem(ctx, constants.RedisParameterDataObjectKeySet, key)
|
|
||||||
}
|
|
||||||
for _, aliasKey := range oldAliasKeys {
|
|
||||||
if _, exists := currentAliasKeys[aliasKey]; exists {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
pipeline.Del(ctx, aliasKey)
|
|
||||||
pipeline.SRem(ctx, constants.RedisParameterDataObjectAliasKeySet, aliasKey)
|
|
||||||
}
|
|
||||||
if len(currentKeys) == 0 {
|
|
||||||
pipeline.Del(ctx, constants.RedisParameterDataObjectKeySet)
|
|
||||||
}
|
|
||||||
if len(currentAliasKeys) == 0 {
|
|
||||||
pipeline.Del(ctx, constants.RedisParameterDataObjectAliasKeySet)
|
|
||||||
}
|
|
||||||
if _, err := pipeline.Exec(ctx); err != nil {
|
|
||||||
return fmt.Errorf("remove stale parameter data-object keys: %w", err)
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
@ -1,124 +0,0 @@
|
||||||
package model
|
|
||||||
|
|
||||||
import (
|
|
||||||
"database/sql"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestBuildParameterDataObjectHashesCreatesCanonicalHashAndLocalAliases(t *testing.T) {
|
|
||||||
records := []ParameterInitializationRecord{
|
|
||||||
{
|
|
||||||
GridTag: "grid000",
|
|
||||||
ZoneTag: "zone000",
|
|
||||||
StationTag: "station000",
|
|
||||||
StationIsLocal: true,
|
|
||||||
ComponentUUID: "component-uuid",
|
|
||||||
ComponentNSPath: "220kV_xuefulu1",
|
|
||||||
ComponentTag: "cable_22",
|
|
||||||
AttributeGroup: "base_extend",
|
|
||||||
AttributeName: "vnom_kv",
|
|
||||||
AttributeValue: "7800.00",
|
|
||||||
AttributeType: "DOUBLE PRECISION",
|
|
||||||
Description: sql.NullString{String: "额定电压", Valid: true},
|
|
||||||
DescriptionCount: 1,
|
|
||||||
DynamicRecordCount: 1,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
hashes, err := buildParameterDataObjectHashes(records)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Len(t, hashes, 1)
|
|
||||||
|
|
||||||
fullToken := "grid000.zone000.station000.220kV_xuefulu1.cable_22.base_extend.vnom_kv"
|
|
||||||
shortToken := "220kV_xuefulu1.cable_22.base_extend.vnom_kv"
|
|
||||||
assert.Equal(t, fullToken, hashes[0].Key)
|
|
||||||
assert.Equal(t, []string{fullToken, shortToken}, hashes[0].Aliases)
|
|
||||||
assert.Equal(t, "7800.00", hashes[0].Fields["value"])
|
|
||||||
assert.Equal(t, "PARAM", hashes[0].Fields["meta"])
|
|
||||||
assert.Equal(t, "DOUBLE PRECISION", hashes[0].Fields["type"])
|
|
||||||
assert.Equal(t, shortToken, hashes[0].Fields["name"])
|
|
||||||
assert.Equal(t, "额定电压", hashes[0].Fields["description"])
|
|
||||||
assert.Equal(t, fullToken, hashes[0].Fields["id"])
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBuildParameterDataObjectHashesSkipsShortKeyForNonLocalStation(t *testing.T) {
|
|
||||||
records := []ParameterInitializationRecord{
|
|
||||||
{
|
|
||||||
GridTag: "grid",
|
|
||||||
ZoneTag: "zone",
|
|
||||||
StationTag: "station",
|
|
||||||
StationIsLocal: false,
|
|
||||||
ComponentUUID: "component-uuid",
|
|
||||||
ComponentNSPath: "nspath",
|
|
||||||
ComponentTag: "component",
|
|
||||||
AttributeGroup: "stable",
|
|
||||||
AttributeName: "attribute",
|
|
||||||
AttributeValue: "true",
|
|
||||||
AttributeType: "BOOLEAN",
|
|
||||||
Description: sql.NullString{String: "属性", Valid: true},
|
|
||||||
DescriptionCount: 1,
|
|
||||||
DynamicRecordCount: 1,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
hashes, err := buildParameterDataObjectHashes(records)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Len(t, hashes, 1)
|
|
||||||
assert.Equal(t, "grid.zone.station.nspath.component.stable.attribute", hashes[0].Key)
|
|
||||||
assert.Equal(t, []string{"grid.zone.station.nspath.component.stable.attribute"}, hashes[0].Aliases)
|
|
||||||
assert.Equal(t, true, hashes[0].Fields["value"])
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBuildParameterDataObjectHashesRejectsAmbiguousShortToken(t *testing.T) {
|
|
||||||
baseRecord := ParameterInitializationRecord{
|
|
||||||
GridTag: "grid1",
|
|
||||||
ZoneTag: "zone1",
|
|
||||||
StationTag: "station1",
|
|
||||||
StationIsLocal: true,
|
|
||||||
ComponentUUID: "component-uuid-1",
|
|
||||||
ComponentNSPath: "nspath",
|
|
||||||
ComponentTag: "component",
|
|
||||||
AttributeGroup: "stable",
|
|
||||||
AttributeName: "attribute",
|
|
||||||
AttributeValue: "1",
|
|
||||||
AttributeType: "INTEGER",
|
|
||||||
Description: sql.NullString{String: "属性", Valid: true},
|
|
||||||
DescriptionCount: 1,
|
|
||||||
DynamicRecordCount: 1,
|
|
||||||
}
|
|
||||||
otherRecord := baseRecord
|
|
||||||
otherRecord.GridTag = "grid2"
|
|
||||||
otherRecord.ZoneTag = "zone2"
|
|
||||||
otherRecord.StationTag = "station2"
|
|
||||||
otherRecord.ComponentUUID = "component-uuid-2"
|
|
||||||
|
|
||||||
_, err := buildParameterDataObjectHashes([]ParameterInitializationRecord{baseRecord, otherRecord})
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "ambiguous parameter token")
|
|
||||||
assert.Contains(t, err.Error(), "nspath.component.stable.attribute")
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestParameterRedisValuePreservesHashRepresentations(t *testing.T) {
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
rawValue string
|
|
||||||
expected any
|
|
||||||
}{
|
|
||||||
{name: "null", rawValue: "null", expected: "null"},
|
|
||||||
{name: "string", rawValue: `"text"`, expected: "text"},
|
|
||||||
{name: "number precision", rawValue: "1234567890.123456789", expected: "1234567890.123456789"},
|
|
||||||
{name: "boolean", rawValue: "true", expected: true},
|
|
||||||
{name: "object", rawValue: `{"key":"value"}`, expected: `{"key":"value"}`},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.name, func(t *testing.T) {
|
|
||||||
actual, err := parameterRedisValue(test.rawValue)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, test.expected, actual)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,343 +0,0 @@
|
||||||
// Package redis provides Redis persistence helpers.
|
|
||||||
package redis
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"fmt"
|
|
||||||
"sort"
|
|
||||||
"strconv"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
redisclient "github.com/redis/go-redis/v9"
|
|
||||||
)
|
|
||||||
|
|
||||||
const redisChangeRestoreTimeout = 5 * time.Second
|
|
||||||
|
|
||||||
type redisHashChange struct {
|
|
||||||
key string
|
|
||||||
field string
|
|
||||||
oldValue string
|
|
||||||
newValue string
|
|
||||||
}
|
|
||||||
|
|
||||||
type redisZSetChange struct {
|
|
||||||
key string
|
|
||||||
oldValues []redisclient.Z
|
|
||||||
newValues []redisclient.Z
|
|
||||||
}
|
|
||||||
|
|
||||||
// RedisChangeSet keeps the Redis changes belonging to one PostgreSQL
|
|
||||||
// transaction. Changes are prepared first and applied together immediately
|
|
||||||
// before the PostgreSQL transaction is committed.
|
|
||||||
type RedisChangeSet struct {
|
|
||||||
client *redisclient.Client
|
|
||||||
hashChanges []redisHashChange
|
|
||||||
zsetChanges []redisZSetChange
|
|
||||||
applied bool
|
|
||||||
}
|
|
||||||
|
|
||||||
func NewRedisChangeSet(client *redisclient.Client) *RedisChangeSet {
|
|
||||||
return &RedisChangeSet{client: client}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) AddHashChange(
|
|
||||||
ctx context.Context,
|
|
||||||
canonicalKey, field string,
|
|
||||||
value any,
|
|
||||||
) error {
|
|
||||||
if changes == nil || changes.client == nil {
|
|
||||||
return fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
if canonicalKey == "" {
|
|
||||||
return fmt.Errorf("canonical redis key is empty")
|
|
||||||
}
|
|
||||||
|
|
||||||
newValue, err := redisChangeString(value)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("encode redis data-object value: %w", err)
|
|
||||||
}
|
|
||||||
oldValue, err := changes.client.HGet(ctx, canonicalKey, field).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query canonical redis hash %q field %q: %w", canonicalKey, field, err)
|
|
||||||
}
|
|
||||||
changes.hashChanges = append(changes.hashChanges, redisHashChange{
|
|
||||||
key: canonicalKey,
|
|
||||||
field: field,
|
|
||||||
oldValue: oldValue,
|
|
||||||
newValue: newValue,
|
|
||||||
})
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) AddMeasurementValueChange(
|
|
||||||
ctx context.Context,
|
|
||||||
key string,
|
|
||||||
value float64,
|
|
||||||
timestamp time.Time,
|
|
||||||
replace bool,
|
|
||||||
) error {
|
|
||||||
if changes == nil || changes.client == nil {
|
|
||||||
return fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
if key == "" {
|
|
||||||
return fmt.Errorf("measurement redis key is empty")
|
|
||||||
}
|
|
||||||
keyType, err := changes.client.Type(ctx, key).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query measurement redis key type for %q: %w", key, err)
|
|
||||||
}
|
|
||||||
if keyType != "none" && keyType != "zset" {
|
|
||||||
return fmt.Errorf("measurement redis key %q has type %q, expected zset", key, keyType)
|
|
||||||
}
|
|
||||||
oldValues, err := changes.client.ZRangeWithScores(ctx, key, 0, -1).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("query measurement redis values for %q: %w", key, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
newMember := strconv.FormatInt(timestamp.UnixNano(), 10)
|
|
||||||
newValues := []redisclient.Z{{Score: value, Member: newMember}}
|
|
||||||
if !replace {
|
|
||||||
newValues = mergeRedisZValues(oldValues, newValues...)
|
|
||||||
}
|
|
||||||
changes.zsetChanges = append(changes.zsetChanges, redisZSetChange{
|
|
||||||
key: key,
|
|
||||||
oldValues: normalizeRedisZValues(oldValues),
|
|
||||||
newValues: normalizeRedisZValues(newValues),
|
|
||||||
})
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) Apply(ctx context.Context) error {
|
|
||||||
if changes == nil || changes.client == nil {
|
|
||||||
return fmt.Errorf("redis client is not initialized")
|
|
||||||
}
|
|
||||||
if len(changes.hashChanges) == 0 && len(changes.zsetChanges) == 0 {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
keys := changes.keys()
|
|
||||||
err := changes.client.Watch(ctx, func(tx *redisclient.Tx) error {
|
|
||||||
if err := changes.verify(ctx, tx, false); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
_, err := tx.TxPipelined(ctx, func(pipe redisclient.Pipeliner) error {
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
pipe.HSet(ctx, change.key, change.field, change.newValue)
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
pipe.Del(ctx, change.key)
|
|
||||||
if len(change.newValues) > 0 {
|
|
||||||
pipe.ZAdd(ctx, change.key, change.newValues...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
return err
|
|
||||||
}, keys...)
|
|
||||||
if err != nil {
|
|
||||||
// A connection error can leave EXEC's outcome unknown. Restore only
|
|
||||||
// when Redis still contains either the prepared or the applied state.
|
|
||||||
restoreCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), redisChangeRestoreTimeout)
|
|
||||||
defer cancel()
|
|
||||||
if restoreErr := changes.restoreAfterApplyFailure(restoreCtx); restoreErr != nil {
|
|
||||||
return fmt.Errorf("apply redis changes: %w; restore redis changes: %v", err, restoreErr)
|
|
||||||
}
|
|
||||||
return fmt.Errorf("apply redis changes: %w", err)
|
|
||||||
}
|
|
||||||
changes.applied = true
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Revert restores Redis after a PostgreSQL commit failure. It uses WATCH and
|
|
||||||
// only restores values that still match this change set.
|
|
||||||
func (changes *RedisChangeSet) Revert(ctx context.Context) error {
|
|
||||||
if changes == nil || !changes.applied {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
return changes.restoreOldValues(ctx, true)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) restoreOldValues(ctx context.Context, compareNew bool) error {
|
|
||||||
keys := changes.keys()
|
|
||||||
return changes.client.Watch(ctx, func(tx *redisclient.Tx) error {
|
|
||||||
if compareNew {
|
|
||||||
if err := changes.verify(ctx, tx, true); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_, err := tx.TxPipelined(ctx, func(pipe redisclient.Pipeliner) error {
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
pipe.HSet(ctx, change.key, change.field, change.oldValue)
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
pipe.Del(ctx, change.key)
|
|
||||||
if len(change.oldValues) > 0 {
|
|
||||||
pipe.ZAdd(ctx, change.key, change.oldValues...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
return err
|
|
||||||
}, keys...)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) restoreAfterApplyFailure(ctx context.Context) error {
|
|
||||||
keys := changes.keys()
|
|
||||||
return changes.client.Watch(ctx, func(tx *redisclient.Tx) error {
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
actual, err := tx.HGet(ctx, change.key, change.field).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("verify redis hash %q field %q after apply failure: %w", change.key, change.field, err)
|
|
||||||
}
|
|
||||||
if actual != change.oldValue && actual != change.newValue {
|
|
||||||
return fmt.Errorf("redis hash %q field %q changed concurrently", change.key, change.field)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
actual, err := tx.ZRangeWithScores(ctx, change.key, 0, -1).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("verify redis zset %q after apply failure: %w", change.key, err)
|
|
||||||
}
|
|
||||||
if !equalRedisZValues(actual, change.oldValues) && !equalRedisZValues(actual, change.newValues) {
|
|
||||||
return fmt.Errorf("redis zset %q changed concurrently", change.key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
_, err := tx.TxPipelined(ctx, func(pipe redisclient.Pipeliner) error {
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
pipe.HSet(ctx, change.key, change.field, change.oldValue)
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
pipe.Del(ctx, change.key)
|
|
||||||
if len(change.oldValues) > 0 {
|
|
||||||
pipe.ZAdd(ctx, change.key, change.oldValues...)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
return err
|
|
||||||
}, keys...)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) verify(ctx context.Context, tx *redisclient.Tx, expectNew bool) error {
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
expected := change.oldValue
|
|
||||||
if expectNew {
|
|
||||||
expected = change.newValue
|
|
||||||
}
|
|
||||||
actual, err := tx.HGet(ctx, change.key, change.field).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("verify redis hash %q field %q: %w", change.key, change.field, err)
|
|
||||||
}
|
|
||||||
if actual != expected {
|
|
||||||
return fmt.Errorf("redis hash %q field %q changed concurrently", change.key, change.field)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
expected := change.oldValues
|
|
||||||
if expectNew {
|
|
||||||
expected = change.newValues
|
|
||||||
}
|
|
||||||
actual, err := tx.ZRangeWithScores(ctx, change.key, 0, -1).Result()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("verify redis zset %q: %w", change.key, err)
|
|
||||||
}
|
|
||||||
if !equalRedisZValues(actual, expected) {
|
|
||||||
return fmt.Errorf("redis zset %q changed concurrently", change.key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (changes *RedisChangeSet) keys() []string {
|
|
||||||
seen := make(map[string]struct{}, len(changes.hashChanges)+len(changes.zsetChanges))
|
|
||||||
keys := make([]string, 0, len(seen))
|
|
||||||
for _, change := range changes.hashChanges {
|
|
||||||
if _, ok := seen[change.key]; !ok {
|
|
||||||
seen[change.key] = struct{}{}
|
|
||||||
keys = append(keys, change.key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for _, change := range changes.zsetChanges {
|
|
||||||
if _, ok := seen[change.key]; !ok {
|
|
||||||
seen[change.key] = struct{}{}
|
|
||||||
keys = append(keys, change.key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
sort.Strings(keys)
|
|
||||||
return keys
|
|
||||||
}
|
|
||||||
|
|
||||||
func redisChangeString(value any) (string, error) {
|
|
||||||
switch typedValue := value.(type) {
|
|
||||||
case string:
|
|
||||||
return typedValue, nil
|
|
||||||
case []byte:
|
|
||||||
return string(typedValue), nil
|
|
||||||
case nil:
|
|
||||||
return "null", nil
|
|
||||||
case bool:
|
|
||||||
return strconv.FormatBool(typedValue), nil
|
|
||||||
case int:
|
|
||||||
return strconv.Itoa(typedValue), nil
|
|
||||||
case int16:
|
|
||||||
return strconv.FormatInt(int64(typedValue), 10), nil
|
|
||||||
case int64:
|
|
||||||
return strconv.FormatInt(typedValue, 10), nil
|
|
||||||
case float64:
|
|
||||||
return strconv.FormatFloat(typedValue, 'f', -1, 64), nil
|
|
||||||
default:
|
|
||||||
encoded, err := json.Marshal(typedValue)
|
|
||||||
if err != nil {
|
|
||||||
return "", err
|
|
||||||
}
|
|
||||||
return string(encoded), nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func cloneRedisZValues(values []redisclient.Z) []redisclient.Z {
|
|
||||||
cloned := make([]redisclient.Z, len(values))
|
|
||||||
copy(cloned, values)
|
|
||||||
return cloned
|
|
||||||
}
|
|
||||||
|
|
||||||
func mergeRedisZValues(current []redisclient.Z, additions ...redisclient.Z) []redisclient.Z {
|
|
||||||
valuesByMember := make(map[string]redisclient.Z, len(current)+len(additions))
|
|
||||||
for _, value := range current {
|
|
||||||
valuesByMember[fmt.Sprint(value.Member)] = value
|
|
||||||
}
|
|
||||||
for _, value := range additions {
|
|
||||||
valuesByMember[fmt.Sprint(value.Member)] = value
|
|
||||||
}
|
|
||||||
values := make([]redisclient.Z, 0, len(valuesByMember))
|
|
||||||
for _, value := range valuesByMember {
|
|
||||||
values = append(values, value)
|
|
||||||
}
|
|
||||||
return normalizeRedisZValues(values)
|
|
||||||
}
|
|
||||||
|
|
||||||
func normalizeRedisZValues(values []redisclient.Z) []redisclient.Z {
|
|
||||||
normalized := cloneRedisZValues(values)
|
|
||||||
sort.Slice(normalized, func(i, j int) bool {
|
|
||||||
if normalized[i].Score != normalized[j].Score {
|
|
||||||
return normalized[i].Score < normalized[j].Score
|
|
||||||
}
|
|
||||||
return fmt.Sprint(normalized[i].Member) < fmt.Sprint(normalized[j].Member)
|
|
||||||
})
|
|
||||||
return normalized
|
|
||||||
}
|
|
||||||
|
|
||||||
func equalRedisZValues(left, right []redisclient.Z) bool {
|
|
||||||
left = normalizeRedisZValues(left)
|
|
||||||
right = normalizeRedisZValues(right)
|
|
||||||
if len(left) != len(right) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
for index := range left {
|
|
||||||
if left[index].Score != right[index].Score ||
|
|
||||||
fmt.Sprint(left[index].Member) != fmt.Sprint(right[index].Member) {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
@ -1,44 +0,0 @@
|
||||||
package redis
|
|
||||||
|
|
||||||
import (
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9"
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestRedisChangeStringUsesCacheRepresentations(t *testing.T) {
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
value any
|
|
||||||
want string
|
|
||||||
}{
|
|
||||||
{name: "string", value: "15.2", want: "15.2"},
|
|
||||||
{name: "integer", value: int64(15), want: "15"},
|
|
||||||
{name: "decimal", value: 15.2, want: "15.2"},
|
|
||||||
{name: "boolean", value: true, want: "true"},
|
|
||||||
{name: "object", value: map[string]any{"enabled": true}, want: `{"enabled":true}`},
|
|
||||||
}
|
|
||||||
for _, test := range tests {
|
|
||||||
t.Run(test.name, func(t *testing.T) {
|
|
||||||
got, err := redisChangeString(test.value)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, test.want, got)
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestMergeRedisZValuesReplacesDuplicateMemberAndSorts(t *testing.T) {
|
|
||||||
values := mergeRedisZValues(
|
|
||||||
[]redis.Z{
|
|
||||||
{Score: 20, Member: "2"},
|
|
||||||
{Score: 10, Member: "1"},
|
|
||||||
},
|
|
||||||
redis.Z{Score: 30, Member: "1"},
|
|
||||||
)
|
|
||||||
assert.Equal(t, []redis.Z{
|
|
||||||
{Score: 20, Member: "2"},
|
|
||||||
{Score: 30, Member: "1"},
|
|
||||||
}, values)
|
|
||||||
}
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
// Package sql defines reusable database SQL statements
|
// Package sql defines reusable database SQL statements.
|
||||||
package sql
|
package sql
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|
|
||||||
|
|
@ -1,38 +0,0 @@
|
||||||
// Package sql defines reusable database SQL statements
|
|
||||||
package sql
|
|
||||||
|
|
||||||
const (
|
|
||||||
// MeasurementInitializationRows joins measurements to the hierarchy used by
|
|
||||||
// all supported data-object token forms. The bay join guarantees that
|
|
||||||
// token6=bay refers to an existing bay record.
|
|
||||||
MeasurementInitializationRows = `SELECT
|
|
||||||
grid.tagname AS grid_tag,
|
|
||||||
zone.tagname AS zone_tag,
|
|
||||||
station.tagname AS station_tag,
|
|
||||||
component.global_uuid::text AS component_uuid,
|
|
||||||
component.nspath AS component_nspath,
|
|
||||||
component.tag AS component_tag,
|
|
||||||
measurement.id AS measurement_id,
|
|
||||||
measurement.tag AS measurement_tag,
|
|
||||||
measurement.name AS measurement_name,
|
|
||||||
measurement.type AS measurement_type,
|
|
||||||
measurement.mode AS measurement_mode,
|
|
||||||
measurement.size AS measurement_size,
|
|
||||||
measurement.data_source AS measurement_data_source,
|
|
||||||
measurement.event_plan AS measurement_event_plan,
|
|
||||||
measurement.binding AS measurement_binding
|
|
||||||
FROM public.grid AS grid
|
|
||||||
INNER JOIN public.zone AS zone ON zone.grid_id = grid.id
|
|
||||||
INNER JOIN public.station AS station ON station.zone_id = zone.id
|
|
||||||
INNER JOIN public.component AS component ON component.station_id = station.id
|
|
||||||
INNER JOIN public.measurement AS measurement
|
|
||||||
ON measurement.component_uuid = component.global_uuid
|
|
||||||
INNER JOIN public.bay AS bay
|
|
||||||
ON bay.bay_uuid = measurement.bay_uuid
|
|
||||||
WHERE grid.tagname <> ''
|
|
||||||
AND zone.tagname <> ''
|
|
||||||
AND station.tagname <> ''
|
|
||||||
AND component.nspath <> ''
|
|
||||||
AND component.tag <> ''
|
|
||||||
AND measurement.tag <> ''`
|
|
||||||
)
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
// Package sql defines reusable database SQL statements
|
// Package sql defines reusable database SQL statements.
|
||||||
package sql
|
package sql
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|
|
||||||
|
|
@ -1,124 +0,0 @@
|
||||||
// Package sql defines reusable database SQL statements
|
|
||||||
package sql
|
|
||||||
|
|
||||||
const (
|
|
||||||
// ParameterInitializationRoutes returns the dynamic-table mappings used by
|
|
||||||
// supported parameter attribute groups.
|
|
||||||
ParameterInitializationRoutes = `SELECT name, tag, group_name
|
|
||||||
FROM project_manager
|
|
||||||
WHERE group_name IN ?`
|
|
||||||
|
|
||||||
// DynamicParameterInitializationRows joins a dynamic parameter table to its
|
|
||||||
// component hierarchy and project_manager route. The table identifier is
|
|
||||||
// inserted only after application-level identifier and allowlist checks.
|
|
||||||
DynamicParameterInitializationRows = `WITH dynamic_rows AS (
|
|
||||||
SELECT dynamic_record.*,
|
|
||||||
COUNT(*) OVER (
|
|
||||||
PARTITION BY dynamic_record.global_uuid, dynamic_record.attribute_group
|
|
||||||
) AS initialization_record_count
|
|
||||||
FROM public.%[1]s AS dynamic_record
|
|
||||||
)
|
|
||||||
SELECT
|
|
||||||
grid.tagname AS grid_tag,
|
|
||||||
zone.tagname AS zone_tag,
|
|
||||||
station.tagname AS station_tag,
|
|
||||||
station.is_local AS station_is_local,
|
|
||||||
component.global_uuid::text AS component_uuid,
|
|
||||||
component.nspath AS component_nspath,
|
|
||||||
component.tag AS component_tag,
|
|
||||||
project.group_name AS attribute_group,
|
|
||||||
attribute.key AS attribute_name,
|
|
||||||
attribute.value::text AS attribute_value,
|
|
||||||
UPPER(pg_catalog.format_type(column_attribute.atttypid, column_attribute.atttypmod)) AS attribute_type,
|
|
||||||
attribute_description.description,
|
|
||||||
attribute_description.description_count,
|
|
||||||
dynamic_row.initialization_record_count AS dynamic_record_count
|
|
||||||
FROM public.grid AS grid
|
|
||||||
INNER JOIN public.zone AS zone ON zone.grid_id = grid.id
|
|
||||||
INNER JOIN public.station AS station ON station.zone_id = zone.id
|
|
||||||
INNER JOIN public.component AS component ON component.station_id = station.id
|
|
||||||
INNER JOIN public.project_manager AS project
|
|
||||||
ON project.tag = component.model_name
|
|
||||||
INNER JOIN dynamic_rows AS dynamic_row
|
|
||||||
ON dynamic_row.global_uuid = component.global_uuid
|
|
||||||
AND dynamic_row.attribute_group = project.group_name
|
|
||||||
CROSS JOIN LATERAL jsonb_each(
|
|
||||||
to_jsonb(dynamic_row)
|
|
||||||
- 'id'
|
|
||||||
- 'global_uuid'
|
|
||||||
- 'attribute_group'
|
|
||||||
- 'initialization_record_count'
|
|
||||||
) AS attribute
|
|
||||||
INNER JOIN pg_catalog.pg_namespace AS table_namespace
|
|
||||||
ON table_namespace.nspname = 'public'
|
|
||||||
INNER JOIN pg_catalog.pg_class AS parameter_table
|
|
||||||
ON parameter_table.relnamespace = table_namespace.oid
|
|
||||||
AND parameter_table.relname = project.name
|
|
||||||
INNER JOIN pg_catalog.pg_attribute AS column_attribute
|
|
||||||
ON column_attribute.attrelid = parameter_table.oid
|
|
||||||
AND column_attribute.attname = attribute.key
|
|
||||||
AND column_attribute.attnum > 0
|
|
||||||
AND NOT column_attribute.attisdropped
|
|
||||||
LEFT JOIN LATERAL (
|
|
||||||
SELECT
|
|
||||||
MIN(basic_attribute.attribute_name) AS description,
|
|
||||||
COUNT(*) AS description_count
|
|
||||||
FROM basic.attribute AS basic_attribute
|
|
||||||
WHERE basic_attribute.attribute = attribute.key
|
|
||||||
) AS attribute_description ON TRUE
|
|
||||||
WHERE project.name = ?
|
|
||||||
AND project.tag = ?
|
|
||||||
AND project.group_name = ?
|
|
||||||
AND grid.tagname <> ''
|
|
||||||
AND zone.tagname <> ''
|
|
||||||
AND station.tagname <> ''
|
|
||||||
AND component.nspath <> ''
|
|
||||||
AND component.tag <> ''`
|
|
||||||
|
|
||||||
// ComponentParameterInitializationRows expands the component table into one
|
|
||||||
// row per queryable component attribute while retaining the full hierarchy.
|
|
||||||
ComponentParameterInitializationRows = `SELECT
|
|
||||||
grid.tagname AS grid_tag,
|
|
||||||
zone.tagname AS zone_tag,
|
|
||||||
station.tagname AS station_tag,
|
|
||||||
station.is_local AS station_is_local,
|
|
||||||
component.global_uuid::text AS component_uuid,
|
|
||||||
component.nspath AS component_nspath,
|
|
||||||
component.tag AS component_tag,
|
|
||||||
'component' AS attribute_group,
|
|
||||||
attribute.key AS attribute_name,
|
|
||||||
attribute.value::text AS attribute_value,
|
|
||||||
UPPER(pg_catalog.format_type(column_attribute.atttypid, column_attribute.atttypmod)) AS attribute_type,
|
|
||||||
attribute_description.description,
|
|
||||||
attribute_description.description_count,
|
|
||||||
1::bigint AS dynamic_record_count
|
|
||||||
FROM public.grid AS grid
|
|
||||||
INNER JOIN public.zone AS zone ON zone.grid_id = grid.id
|
|
||||||
INNER JOIN public.station AS station ON station.zone_id = zone.id
|
|
||||||
INNER JOIN public.component AS component ON component.station_id = station.id
|
|
||||||
CROSS JOIN LATERAL jsonb_each(
|
|
||||||
to_jsonb(component) - 'station_id'
|
|
||||||
) AS attribute
|
|
||||||
INNER JOIN pg_catalog.pg_namespace AS table_namespace
|
|
||||||
ON table_namespace.nspname = 'public'
|
|
||||||
INNER JOIN pg_catalog.pg_class AS component_table
|
|
||||||
ON component_table.relnamespace = table_namespace.oid
|
|
||||||
AND component_table.relname = 'component'
|
|
||||||
INNER JOIN pg_catalog.pg_attribute AS column_attribute
|
|
||||||
ON column_attribute.attrelid = component_table.oid
|
|
||||||
AND column_attribute.attname = attribute.key
|
|
||||||
AND column_attribute.attnum > 0
|
|
||||||
AND NOT column_attribute.attisdropped
|
|
||||||
LEFT JOIN LATERAL (
|
|
||||||
SELECT
|
|
||||||
MIN(basic_attribute.attribute_name) AS description,
|
|
||||||
COUNT(*) AS description_count
|
|
||||||
FROM basic.attribute AS basic_attribute
|
|
||||||
WHERE basic_attribute.attribute = attribute.key
|
|
||||||
) AS attribute_description ON TRUE
|
|
||||||
WHERE grid.tagname <> ''
|
|
||||||
AND zone.tagname <> ''
|
|
||||||
AND station.tagname <> ''
|
|
||||||
AND component.nspath <> ''
|
|
||||||
AND component.tag <> ''`
|
|
||||||
)
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
// Package sql defines reusable database SQL statements
|
// Package sql define database sql statement
|
||||||
package sql
|
package sql
|
||||||
|
|
||||||
// RecursiveSQL define topologic table recursive query statement
|
// RecursiveSQL define topologic table recursive query statement
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue