diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a1e5868..ce60bae 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -81,13 +81,13 @@ jobs: cache-dependency-path: go.sum - name: Test Runtime B - run: go test -count=1 ./runtime-b/... + run: go test -count=1 ./runtime-b/... ./internal/nacosregistration/... - name: Race test Runtime B - run: go test -race ./runtime-b/... + run: go test -race ./runtime-b/... ./internal/nacosregistration/... - name: Vet Runtime B - run: go vet ./runtime-b/... + run: go vet ./runtime-b/... ./internal/nacosregistration/... images: runs-on: ubuntu-latest diff --git a/go.mod b/go.mod index fb904e3..ad96afd 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/NeKiro-project/NeKiro-Samples go 1.26.0 require ( - github.com/NeKiro-project/NeKiro v0.0.0-20260804142931-aad73c450435 + github.com/NeKiro-project/NeKiro v0.0.0-20260810043416-3e815b89cb37 github.com/NeKiro-project/nekiro-sdk-go v0.0.0-20260804145402-39c37c8929b5 github.com/a2aproject/a2a-go v0.3.15 github.com/golang-jwt/jwt/v5 v5.3.1 @@ -17,6 +17,7 @@ require ( github.com/go-logr/stdr v1.2.2 // indirect github.com/google/uuid v1.6.0 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect + github.com/nacos-group/nacos-sdk-go/v2 v2.3.5 // indirect github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect go.opentelemetry.io/auto/sdk v1.1.0 // indirect go.opentelemetry.io/otel v1.35.0 // indirect diff --git a/go.sum b/go.sum index 093208f..932020b 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ github.com/Masterminds/semver/v3 v3.5.0 h1:kQceYJfbupGfZOKZQg0kou0DgAKhzDg2NZPAwZ/2OOE= github.com/Masterminds/semver/v3 v3.5.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= -github.com/NeKiro-project/NeKiro v0.0.0-20260804142931-aad73c450435 h1:Qwd8L31Yto8HZgDvekdMi53jEMNAt1uwqiXPh8IWkVA= -github.com/NeKiro-project/NeKiro v0.0.0-20260804142931-aad73c450435/go.mod h1:ceTCf86U3pNqSjqJXWgkOhfCv+BbEUicGD1IfY+R8Js= +github.com/NeKiro-project/NeKiro v0.0.0-20260810043416-3e815b89cb37 h1:ai0eN+G2k6rwtF9h1dDhDKzloueH6AyuJp2UOIoyQec= +github.com/NeKiro-project/NeKiro v0.0.0-20260810043416-3e815b89cb37/go.mod h1:JCIEeiLu52WC/Q5QlcAKmWKRtW7CNLkZ3lV3BAn92Oo= github.com/NeKiro-project/nekiro-sdk-go v0.0.0-20260804145402-39c37c8929b5 h1:wx8nuBluyNb7MhMOURtX/9/Q+FBNcniJqoCMpxBnrwE= github.com/NeKiro-project/nekiro-sdk-go v0.0.0-20260804145402-39c37c8929b5/go.mod h1:lxAQLsSVXE3lmWoz+SMDKsDOc1b0Vj7okAGNkE7KFRI= github.com/a2aproject/a2a-go v0.3.15 h1:h5YpCiPq3jxQ5rIns7oDjPag3ivP8u817AzdA4F+NiI= @@ -46,6 +46,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuErjs= github.com/mattn/go-sqlite3 v1.14.32/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/nacos-group/nacos-sdk-go/v2 v2.3.5 h1:Hux7C4N4rWhwBF5Zm4yyYskrs9VTgrRTA8DZjoEhQTs= +github.com/nacos-group/nacos-sdk-go/v2 v2.3.5/go.mod h1:ygUBdt7eGeYBt6Lz2HO3wx7crKXk25Mp80568emGMWU= github.com/oasdiff/yaml v0.1.1 h1:6nHx+pn9gBRM6YpBlFZFQGCCd1nuvqOBtTD3KKTgGxY= github.com/oasdiff/yaml v0.1.1/go.mod h1:EYJNoyktvWMJ0Hmhx+6qTaqMOsalUaRGT8Sj1hNcegU= github.com/oasdiff/yaml3 v0.0.14 h1:aLJee3hxBK2H5wdXd9iPcIXb93Nty1Ge0pT171eHtkw= diff --git a/internal/nacosregistration/config.go b/internal/nacosregistration/config.go index f905326..c824703 100644 --- a/internal/nacosregistration/config.go +++ b/internal/nacosregistration/config.go @@ -8,6 +8,8 @@ import ( "strconv" "strings" "time" + + "github.com/NeKiro-project/NeKiro/registry" ) const ( @@ -23,22 +25,32 @@ var identifierPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$` type Config struct { Mode string + AgentID string + InstanceID string + AgentCardVersion string + ReleaseID string + CardDigest string + CanonicalEndpoint string + Audience string APIOrigin string NamespaceID string GroupName string ServiceName string ClusterName string + PortName string AdvertisedIP string AdvertisedPort int + Weight float64 HeartbeatInterval time.Duration + HeartbeatTimeout time.Duration + IPDeleteTimeout time.Duration RequestTimeout time.Duration AuthMode string AccessToken string - InstanceID string } -func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config, error) { - if lookup == nil || !identifierPattern.MatchString(instanceID) || prefix != "RUNTIME_A" && prefix != "RUNTIME_B" { +func Load(lookup func(string) (string, bool), prefix, agentID, instanceID string) (Config, error) { + if lookup == nil || !identifierPattern.MatchString(agentID) || !identifierPattern.MatchString(instanceID) || prefix != "RUNTIME_A" && prefix != "RUNTIME_B" { return Config{}, errorsFor(prefix, "registration dependencies are invalid") } name := func(suffix string) string { return prefix + "_" + suffix } @@ -46,8 +58,13 @@ func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config if err != nil { return Config{}, err } - config := Config{Mode: mode, InstanceID: instanceID} - nacosSuffixes := []string{"NACOS_API_ORIGIN", "NACOS_NAMESPACE_ID", "NACOS_GROUP_NAME", "NACOS_SERVICE_NAME", "NACOS_CLUSTER_NAME", "NACOS_ADVERTISED_IP", "NACOS_ADVERTISED_PORT", "NACOS_HEARTBEAT_INTERVAL_MS", "NACOS_REQUEST_TIMEOUT_MS", "NACOS_AUTH_MODE", "NACOS_ACCESS_TOKEN"} + config := Config{Mode: mode, AgentID: agentID, InstanceID: instanceID} + nacosSuffixes := []string{ + "AGENT_CARD_VERSION", "RELEASE_ID", "CARD_DIGEST", "CANONICAL_ENDPOINT", "AUDIENCE", + "NACOS_API_ORIGIN", "NACOS_NAMESPACE_ID", "NACOS_GROUP_NAME", "NACOS_SERVICE_NAME", "NACOS_CLUSTER_NAME", "NACOS_PORT_NAME", + "NACOS_ADVERTISED_IP", "NACOS_ADVERTISED_PORT", "NACOS_WEIGHT", "NACOS_HEARTBEAT_INTERVAL_MS", "NACOS_HEARTBEAT_TIMEOUT_MS", + "NACOS_IP_DELETE_TIMEOUT_MS", "NACOS_REQUEST_TIMEOUT_MS", "NACOS_AUTH_MODE", "NACOS_ACCESS_TOKEN", + } if mode == ModeDisabled { for _, suffix := range nacosSuffixes { if _, exists := lookup(name(suffix)); exists { @@ -59,6 +76,18 @@ func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config if mode != ModeNacos { return Config{}, fmt.Errorf("%s is unsupported", name("REGISTRATION_MODE")) } + for environment, destination := range map[string]*string{ + name("AGENT_CARD_VERSION"): &config.AgentCardVersion, + name("RELEASE_ID"): &config.ReleaseID, + name("CARD_DIGEST"): &config.CardDigest, + name("CANONICAL_ENDPOINT"): &config.CanonicalEndpoint, + name("AUDIENCE"): &config.Audience, + } { + *destination, err = required(lookup, environment) + if err != nil { + return Config{}, err + } + } if config.APIOrigin, err = required(lookup, name("NACOS_API_ORIGIN")); err != nil { return Config{}, err } @@ -70,6 +99,7 @@ func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config name("NACOS_GROUP_NAME"): &config.GroupName, name("NACOS_SERVICE_NAME"): &config.ServiceName, name("NACOS_CLUSTER_NAME"): &config.ClusterName, + name("NACOS_PORT_NAME"): &config.PortName, } { *destination, err = requiredIdentifier(lookup, environment) if err != nil { @@ -86,16 +116,23 @@ func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config return Config{}, err } config.AdvertisedPort = int(port) - heartbeat, err := requiredUnsigned(lookup, name("NACOS_HEARTBEAT_INTERVAL_MS"), minimumMillis, maximumMillis) + weight, err := requiredUnsigned(lookup, name("NACOS_WEIGHT"), 1, 10000) if err != nil { return Config{}, err } - config.HeartbeatInterval = time.Duration(heartbeat) * time.Millisecond - timeout, err := requiredUnsigned(lookup, name("NACOS_REQUEST_TIMEOUT_MS"), minimumMillis, maximumMillis) - if err != nil { + config.Weight = float64(weight) + if config.HeartbeatInterval, err = requiredDuration(lookup, name("NACOS_HEARTBEAT_INTERVAL_MS"), 1000, maximumMillis); err != nil { + return Config{}, err + } + if config.HeartbeatTimeout, err = requiredDuration(lookup, name("NACOS_HEARTBEAT_TIMEOUT_MS"), 1001, 300000); err != nil { + return Config{}, err + } + if config.IPDeleteTimeout, err = requiredDuration(lookup, name("NACOS_IP_DELETE_TIMEOUT_MS"), 1002, 600000); err != nil { + return Config{}, err + } + if config.RequestTimeout, err = requiredDuration(lookup, name("NACOS_REQUEST_TIMEOUT_MS"), minimumMillis, maximumMillis); err != nil { return Config{}, err } - config.RequestTimeout = time.Duration(timeout) * time.Millisecond config.AuthMode, err = required(lookup, name("NACOS_AUTH_MODE")) if err != nil { return Config{}, err @@ -117,17 +154,23 @@ func Load(lookup func(string) (string, bool), prefix, instanceID string) (Config } func (config Config) Validate() error { - if !identifierPattern.MatchString(config.InstanceID) { - return errorsFor("runtime", "instance ID is invalid") + if !identifierPattern.MatchString(config.AgentID) || !identifierPattern.MatchString(config.InstanceID) { + return errorsFor("runtime", "registration identity is invalid") } if config.Mode == ModeDisabled { return nil } - if config.Mode != ModeNacos || validateOrigin(config.APIOrigin, "Nacos API origin") != nil || !identifierPattern.MatchString(config.NamespaceID) || !identifierPattern.MatchString(config.GroupName) || !identifierPattern.MatchString(config.ServiceName) || !identifierPattern.MatchString(config.ClusterName) { + if config.Mode != ModeNacos || validateOrigin(config.APIOrigin, "Nacos API origin") != nil || !identifierPattern.MatchString(config.NamespaceID) || !identifierPattern.MatchString(config.GroupName) || !identifierPattern.MatchString(config.ServiceName) || !identifierPattern.MatchString(config.ClusterName) || !identifierPattern.MatchString(config.PortName) { return errorsFor("runtime", "Nacos registration tuple is invalid") } + if _, err := registry.NewReleaseTarget(registry.ReleaseTargetInput{ + AgentID: config.AgentID, AgentCardVersion: config.AgentCardVersion, ReleaseID: config.ReleaseID, + CardDigest: config.CardDigest, CanonicalEndpoint: config.CanonicalEndpoint, Audience: config.Audience, + }); err != nil { + return errorsFor("runtime", "exact Release target is invalid") + } parsedIP := net.ParseIP(config.AdvertisedIP) - if parsedIP == nil || parsedIP.String() != config.AdvertisedIP || config.AdvertisedPort < 1 || config.AdvertisedPort > 65535 || config.HeartbeatInterval < minimumMillis*time.Millisecond || config.HeartbeatInterval > maximumMillis*time.Millisecond || config.RequestTimeout < minimumMillis*time.Millisecond || config.RequestTimeout > maximumMillis*time.Millisecond { + if parsedIP == nil || parsedIP.String() != config.AdvertisedIP || config.AdvertisedPort < 1 || config.AdvertisedPort > 65535 || config.Weight < 1 || config.Weight > 10000 || config.Weight != float64(int(config.Weight)) || config.HeartbeatInterval < time.Second || config.HeartbeatInterval > time.Minute || config.HeartbeatTimeout <= config.HeartbeatInterval || config.HeartbeatTimeout > 5*time.Minute || config.IPDeleteTimeout <= config.HeartbeatTimeout || config.IPDeleteTimeout > 10*time.Minute || config.RequestTimeout < minimumMillis*time.Millisecond || config.RequestTimeout > maximumMillis*time.Millisecond { return errorsFor("runtime", "Nacos registration endpoint or timing is invalid") } if config.AuthMode != AuthNone && config.AuthMode != AuthAccessToken || config.AuthMode == AuthNone && config.AccessToken != "" || config.AuthMode == AuthAccessToken && strings.TrimSpace(config.AccessToken) == "" { @@ -169,6 +212,11 @@ func requiredUnsigned(lookup func(string) (string, bool), name string, minimum, return parsed, nil } +func requiredDuration(lookup func(string) (string, bool), name string, minimum, maximum int64) (time.Duration, error) { + value, err := requiredUnsigned(lookup, name, minimum, maximum) + return time.Duration(value) * time.Millisecond, err +} + func validateOrigin(value, name string) error { parsed, err := url.Parse(value) if err != nil || parsed.Scheme != "http" && parsed.Scheme != "https" || parsed.Host == "" || parsed.User != nil || parsed.Path != "/nacos" || parsed.RawPath != "" || parsed.RawQuery != "" || parsed.ForceQuery || parsed.Fragment != "" || parsed.RawFragment != "" { diff --git a/internal/nacosregistration/config_test.go b/internal/nacosregistration/config_test.go new file mode 100644 index 0000000..5d29d66 --- /dev/null +++ b/internal/nacosregistration/config_test.go @@ -0,0 +1,60 @@ +package nacosregistration + +import "testing" + +func TestLoadRequiresExactReleaseAndExplicitFreshness(t *testing.T) { + values := validEnvironment() + config, err := Load(mapLookup(values), "RUNTIME_B", "runtime-b", "runtime-b-primary") + if err != nil { + t.Fatal(err) + } + if config.ReleaseID != "rel_runtime_b_1" || config.PortName != "a2a" || config.HeartbeatTimeout.Milliseconds() != 5000 || config.IPDeleteTimeout.Milliseconds() != 10000 { + t.Fatalf("config=%#v", config) + } + for _, name := range []string{ + "RUNTIME_B_RELEASE_ID", "RUNTIME_B_CARD_DIGEST", "RUNTIME_B_CANONICAL_ENDPOINT", "RUNTIME_B_AUDIENCE", + "RUNTIME_B_NACOS_PORT_NAME", "RUNTIME_B_NACOS_WEIGHT", "RUNTIME_B_NACOS_HEARTBEAT_TIMEOUT_MS", "RUNTIME_B_NACOS_IP_DELETE_TIMEOUT_MS", + } { + invalid := validEnvironment() + delete(invalid, name) + if _, err := Load(mapLookup(invalid), "RUNTIME_B", "runtime-b", "runtime-b-primary"); err == nil { + t.Errorf("missing %s was accepted", name) + } + } +} + +func TestLoadRejectsMismatchedTargetAndFreshnessOrder(t *testing.T) { + for name, mutate := range map[string]func(map[string]string){ + "audience": func(values map[string]string) { values["RUNTIME_B_AUDIENCE"] = "http://runtime-a:8091" }, + "heartbeat timeout": func(values map[string]string) { values["RUNTIME_B_NACOS_HEARTBEAT_TIMEOUT_MS"] = "1000" }, + "delete timeout": func(values map[string]string) { values["RUNTIME_B_NACOS_IP_DELETE_TIMEOUT_MS"] = "5000" }, + } { + t.Run(name, func(t *testing.T) { + values := validEnvironment() + mutate(values) + if _, err := Load(mapLookup(values), "RUNTIME_B", "runtime-b", "runtime-b-primary"); err == nil { + t.Fatal("invalid registration config was accepted") + } + }) + } +} + +func validEnvironment() map[string]string { + return map[string]string{ + "RUNTIME_B_REGISTRATION_MODE": "nacos", "RUNTIME_B_AGENT_CARD_VERSION": "1.0.0", "RUNTIME_B_RELEASE_ID": "rel_runtime_b_1", + "RUNTIME_B_CARD_DIGEST": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "RUNTIME_B_CANONICAL_ENDPOINT": "http://runtime-b:8092/", "RUNTIME_B_AUDIENCE": "http://runtime-b:8092", + "RUNTIME_B_NACOS_API_ORIGIN": "http://nacos:8848/nacos", "RUNTIME_B_NACOS_NAMESPACE_ID": "public", + "RUNTIME_B_NACOS_GROUP_NAME": "NEKIRO", "RUNTIME_B_NACOS_SERVICE_NAME": "runtime-b", "RUNTIME_B_NACOS_CLUSTER_NAME": "DEFAULT", + "RUNTIME_B_NACOS_PORT_NAME": "a2a", "RUNTIME_B_NACOS_ADVERTISED_IP": "127.0.0.1", "RUNTIME_B_NACOS_ADVERTISED_PORT": "8092", + "RUNTIME_B_NACOS_WEIGHT": "1", "RUNTIME_B_NACOS_HEARTBEAT_INTERVAL_MS": "1000", "RUNTIME_B_NACOS_HEARTBEAT_TIMEOUT_MS": "5000", + "RUNTIME_B_NACOS_IP_DELETE_TIMEOUT_MS": "10000", "RUNTIME_B_NACOS_REQUEST_TIMEOUT_MS": "1000", "RUNTIME_B_NACOS_AUTH_MODE": "none", + } +} + +func mapLookup(values map[string]string) func(string) (string, bool) { + return func(name string) (string, bool) { + value, ok := values[name] + return value, ok + } +} diff --git a/internal/nacosregistration/registration.go b/internal/nacosregistration/registration.go index 42baafd..2a9ea94 100644 --- a/internal/nacosregistration/registration.go +++ b/internal/nacosregistration/registration.go @@ -2,199 +2,129 @@ package nacosregistration import ( "context" - "encoding/json" "errors" "fmt" - "io" - "net/http" - "net/url" - "strconv" - "strings" + "net" "sync" - "time" -) -const responseLimitBytes = 4096 + "github.com/NeKiro-project/NeKiro/registry" + registrynacos "github.com/NeKiro-project/NeKiro/registry/nacos" +) -type HTTPDoer interface { - Do(*http.Request) (*http.Response, error) -} +type HTTPDoer = registrynacos.RequestExecutor type Registration struct { - config Config - executor HTTPDoer - mu sync.RWMutex - registered bool - ready bool + registrar registry.InstanceRegistrar + registration registry.Registration + + mu sync.RWMutex + lease registry.InstanceLease } func New(config Config, executor HTTPDoer) (*Registration, error) { if config.Mode != ModeNacos || executor == nil || config.Validate() != nil { return nil, errors.New("Nacos registration dependencies are invalid") } - return &Registration{config: config, executor: executor}, nil + target, _ := registry.NewReleaseTarget(registry.ReleaseTargetInput{ + AgentID: config.AgentID, AgentCardVersion: config.AgentCardVersion, ReleaseID: config.ReleaseID, + CardDigest: config.CardDigest, CanonicalEndpoint: config.CanonicalEndpoint, Audience: config.Audience, + }) + binding, err := registrynacos.NewBinding(registrynacos.BindingInput{ + Target: target, ServiceName: config.ServiceName, GroupName: config.GroupName, ClusterName: config.ClusterName, + }) + if err != nil { + return nil, errors.New("Nacos registration binding is invalid") + } + addressType := registry.AddressTypeIPv4 + if parsed := net.ParseIP(config.AdvertisedIP); parsed != nil && parsed.To4() == nil { + addressType = registry.AddressTypeIPv6 + } + endpoint, err := registry.NewNetworkEndpoint(registry.NetworkEndpointInput{ + AddressType: addressType, Address: config.AdvertisedIP, PortName: config.PortName, + Port: config.AdvertisedPort, Protocol: registry.TransportProtocolTCP, + }) + if err != nil { + return nil, errors.New("Nacos registration endpoint is invalid") + } + weight := int(config.Weight) + instance, err := registry.NewInstance(registry.InstanceInput{ + ID: config.InstanceID, Endpoints: []registry.NetworkEndpoint{endpoint}, Ready: true, Serving: true, Weight: &weight, + }) + if err != nil { + return nil, errors.New("Nacos registration instance is invalid") + } + registration, err := registry.NewRegistration(registry.RegistrationInput{Target: target, Instance: instance}) + if err != nil { + return nil, errors.New("Nacos registration is invalid") + } + registrar, err := registrynacos.NewRegistrar(registrynacos.RegistrarConfig{ + APIOrigin: config.APIOrigin, NamespaceID: config.NamespaceID, Binding: binding, PortName: config.PortName, + Weight: config.Weight, HeartbeatInterval: config.HeartbeatInterval, HeartbeatTimeout: config.HeartbeatTimeout, + IPDeleteTimeout: config.IPDeleteTimeout, AuthMode: config.AuthMode, AccessToken: config.AccessToken, Executor: executor, + }) + if err != nil { + return nil, errors.New("Nacos registrar configuration is invalid") + } + return &Registration{registrar: registrar, registration: registration}, nil } func (value *Registration) Register(ctx context.Context) error { - metadata, _ := json.Marshal(map[string]string{"nekiro.instanceId": value.config.InstanceID}) - parameters := value.baseValues() - parameters.Set("ephemeral", "true") - parameters.Set("enabled", "true") - parameters.Set("healthy", "true") - parameters.Set("weight", "1.0") - parameters.Set("metadata", string(metadata)) - if err := value.request(ctx, http.MethodPost, parameters); err != nil { - value.setState(false, false) + value.mu.Lock() + defer value.mu.Unlock() + if value.lease != nil { + return errors.New("Nacos registration has already started") + } + lease, err := value.registrar.Register(ctx, value.registration) + if err != nil { return fmt.Errorf("register runtime with Nacos: %w", err) } - value.setState(true, true) + value.lease = lease return nil } func (value *Registration) Run(ctx context.Context) error { + if ctx == nil { + return errors.New("Nacos registration context is required") + } value.mu.RLock() - registered := value.registered + lease := value.lease value.mu.RUnlock() - if !registered { + if lease == nil { return errors.New("Nacos registration has not started") } - ticker := time.NewTicker(value.config.HeartbeatInterval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - value.setReady(false) - return nil - case <-ticker.C: - if err := value.Heartbeat(ctx); err != nil { - value.setReady(false) - return fmt.Errorf("heartbeat Nacos registration: %w", err) - } - } + select { + case <-ctx.Done(): + return nil + case <-lease.Done(): + return fmt.Errorf("Nacos registration lease terminated: %w", lease.Err()) } } -func (value *Registration) Heartbeat(ctx context.Context) error { - beat, _ := json.Marshal(struct { - IP string `json:"ip"` - Port int `json:"port"` - Service string `json:"serviceName"` - Cluster string `json:"cluster"` - Metadata map[string]string `json:"metadata"` - Weight float64 `json:"weight"` - Scheduled bool `json:"scheduled"` - }{ - IP: value.config.AdvertisedIP, Port: value.config.AdvertisedPort, - Service: value.config.GroupName + "@@" + value.config.ServiceName, Cluster: value.config.ClusterName, - Metadata: map[string]string{"nekiro.instanceId": value.config.InstanceID}, - Weight: 1.0, Scheduled: false, - }) - parameters := value.baseValues() - parameters.Set("beat", string(beat)) - return value.request(ctx, http.MethodPut, parameters) -} - func (value *Registration) Deregister(ctx context.Context) error { + if ctx == nil { + return errors.New("Nacos deregistration context is required") + } value.mu.RLock() - registered := value.registered + lease := value.lease value.mu.RUnlock() - value.setReady(false) - if !registered { - return nil - } - err := value.request(ctx, http.MethodDelete, value.baseValues()) - value.setState(false, false) - if err != nil { - return fmt.Errorf("deregister runtime from Nacos: %w", err) + var leaseErr error + if lease != nil { + leaseErr = lease.Close(ctx) } - return nil + return errors.Join(leaseErr, value.registrar.Close()) } func (value *Registration) Ready() bool { value.mu.RLock() - defer value.mu.RUnlock() - return value.ready -} - -func (value *Registration) baseValues() url.Values { - parameters := url.Values{ - "serviceName": {value.config.ServiceName}, - "groupName": {value.config.GroupName}, - "clusterName": {value.config.ClusterName}, - "namespaceId": {value.config.NamespaceID}, - "ip": {value.config.AdvertisedIP}, - "port": {strconv.Itoa(value.config.AdvertisedPort)}, - "ephemeral": {"true"}, - } - if value.config.AuthMode == AuthAccessToken { - parameters.Set("accessToken", value.config.AccessToken) - } - return parameters -} - -func (value *Registration) request(ctx context.Context, method string, parameters url.Values) error { - if ctx == nil { - return errors.New("Nacos request context is required") - } - requestContext, cancel := context.WithTimeout(ctx, value.config.RequestTimeout) - defer cancel() - endpoint := strings.TrimSuffix(value.config.APIOrigin, "/") + "/v1/ns/instance" - if method == http.MethodPut { - endpoint += "/beat" - } - var body io.Reader - if method == http.MethodPut { - body = strings.NewReader(parameters.Encode()) - } else { - endpoint += "?" + parameters.Encode() - } - request, err := http.NewRequestWithContext(requestContext, method, endpoint, body) - if err != nil { - return errors.New("build Nacos request") - } - if method == http.MethodPut { - request.Header.Set("Content-Type", "application/x-www-form-urlencoded") - } - response, err := value.executor.Do(request) - if err != nil { - if requestContext.Err() != nil { - return errors.New("Nacos request canceled or timed out") - } - return errors.New("Nacos request unavailable") - } - defer response.Body.Close() - responseBody, err := io.ReadAll(io.LimitReader(response.Body, responseLimitBytes+1)) - if err != nil { - return errors.New("read Nacos response") - } - if len(responseBody) > responseLimitBytes { - return errors.New("Nacos response exceeds limit") - } - if response.StatusCode != http.StatusOK { - return fmt.Errorf("Nacos returned status %d", response.StatusCode) + lease := value.lease + value.mu.RUnlock() + if lease == nil { + return false } - if method == http.MethodPut { - var outcome struct { - Code int `json:"code"` - } - if json.Unmarshal(responseBody, &outcome) != nil || outcome.Code != 10200 { - return errors.New("Nacos heartbeat outcome is invalid") - } - } else if strings.TrimSpace(string(responseBody)) != "ok" { - return errors.New("Nacos registration outcome is invalid") + select { + case <-lease.Done(): + return false + default: + return true } - return nil -} - -func (value *Registration) setReady(ready bool) { - value.mu.Lock() - value.ready = ready - value.mu.Unlock() -} - -func (value *Registration) setState(registered, ready bool) { - value.mu.Lock() - value.registered = registered - value.ready = ready - value.mu.Unlock() } diff --git a/internal/nacosregistration/registration_test.go b/internal/nacosregistration/registration_test.go new file mode 100644 index 0000000..8e7937e --- /dev/null +++ b/internal/nacosregistration/registration_test.go @@ -0,0 +1,71 @@ +package nacosregistration + +import ( + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func TestRegistrationUsesCoreLeaseAndDeregisters(t *testing.T) { + var deletes atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Method == http.MethodDelete { + deletes.Add(1) + } + _, _ = writer.Write([]byte("ok")) + })) + t.Cleanup(server.Close) + registration, err := New(validConfig(server.URL+"/nacos"), server.Client()) + if err != nil { + t.Fatal(err) + } + if registration.Ready() { + t.Fatal("registration was ready before publish") + } + if err := registration.Register(t.Context()); err != nil || !registration.Ready() { + t.Fatalf("register ready=%v error=%v", registration.Ready(), err) + } + if err := registration.Deregister(t.Context()); err != nil || registration.Ready() { + t.Fatalf("deregister ready=%v error=%v", registration.Ready(), err) + } + if err := registration.Deregister(t.Context()); err != nil || deletes.Load() != 1 { + t.Fatalf("idempotent deregister deletes=%d error=%v", deletes.Load(), err) + } +} + +func TestRegistrationLeaseFailureIsTerminal(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Method == http.MethodPut { + writer.WriteHeader(http.StatusServiceUnavailable) + return + } + _, _ = writer.Write([]byte("ok")) + })) + t.Cleanup(server.Close) + registration, err := New(validConfig(server.URL+"/nacos"), server.Client()) + if err != nil { + t.Fatal(err) + } + if err := registration.Register(t.Context()); err != nil { + t.Fatal(err) + } + if err := registration.Run(t.Context()); err == nil || registration.Ready() { + t.Fatalf("terminal lease error=%v ready=%v", err, registration.Ready()) + } + if err := registration.Deregister(t.Context()); err != nil { + t.Fatal(err) + } +} + +func validConfig(origin string) Config { + return Config{ + Mode: ModeNacos, AgentID: "runtime-b", InstanceID: "runtime-b-primary", AgentCardVersion: "1.0.0", ReleaseID: "rel_runtime_b_1", + CardDigest: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + CanonicalEndpoint: "http://runtime-b:8092/", Audience: "http://runtime-b:8092", APIOrigin: origin, + NamespaceID: "public", GroupName: "NEKIRO", ServiceName: "runtime-b", ClusterName: "DEFAULT", PortName: "a2a", + AdvertisedIP: "127.0.0.1", AdvertisedPort: 8092, Weight: 1, HeartbeatInterval: time.Second, + HeartbeatTimeout: 5 * time.Second, IPDeleteTimeout: 10 * time.Second, RequestTimeout: time.Second, AuthMode: AuthNone, + } +} diff --git a/runtime-a/README.md b/runtime-a/README.md index 864e36b..bec70d3 100644 --- a/runtime-a/README.md +++ b/runtime-a/README.md @@ -26,6 +26,16 @@ NEKIRO_AGENT_ROUTER_KEY_ID NEKIRO_AGENT_ROUTER_PUBLIC_KEY_BASE64URL ``` +When `RUNTIME_A_REGISTRATION_MODE=nacos`, the deployment must additionally +provide the exact target fields `RUNTIME_A_AGENT_CARD_VERSION`, +`RUNTIME_A_RELEASE_ID`, `RUNTIME_A_CARD_DIGEST`, +`RUNTIME_A_CANONICAL_ENDPOINT`, and `RUNTIME_A_AUDIENCE`; the Nacos tuple; +`RUNTIME_A_NACOS_PORT_NAME`, advertised IP/port and weight; explicit heartbeat, +heartbeat-timeout, IP-delete-timeout, and request-timeout values; and the +selected authentication mode. Runtime A uses Core's `InstanceRegistrar` and +`InstanceLease`, fails startup if the initial publish fails, becomes not-ready +and stops on terminal lease failure, and explicitly deregisters on shutdown. + `NEKIRO_AGENT_CHALLENGE_DIRECTORY` is an absolute, explicitly configured directory used only to serve provider-owned one-time HTTP ownership proofs at `/.well-known/nekiro/challenges/{challengeId}`. It has no default and is not a diff --git a/runtime-a/cmd/runtime-a/main.go b/runtime-a/cmd/runtime-a/main.go index 7387c47..ea9912c 100644 --- a/runtime-a/cmd/runtime-a/main.go +++ b/runtime-a/cmd/runtime-a/main.go @@ -27,7 +27,7 @@ func run() error { if err != nil { return err } - registrationConfig, err := nacosregistration.Load(os.LookupEnv, "RUNTIME_A", config.InstanceID) + registrationConfig, err := nacosregistration.Load(os.LookupEnv, "RUNTIME_A", config.AgentID, config.InstanceID) if err != nil { return err } diff --git a/runtime-b/README.md b/runtime-b/README.md index 6bbfbfd..f92b925 100644 --- a/runtime-b/README.md +++ b/runtime-b/README.md @@ -30,14 +30,23 @@ NEKIRO_AGENT_ROUTER_PUBLIC_KEY_BASE64URL # Explicit registration lifecycle: set mode to `disabled` for a static sample, # or set mode to `nacos` and provide every Nacos variable below. RUNTIME_B_REGISTRATION_MODE +RUNTIME_B_AGENT_CARD_VERSION +RUNTIME_B_RELEASE_ID +RUNTIME_B_CARD_DIGEST +RUNTIME_B_CANONICAL_ENDPOINT +RUNTIME_B_AUDIENCE RUNTIME_B_NACOS_API_ORIGIN RUNTIME_B_NACOS_NAMESPACE_ID RUNTIME_B_NACOS_GROUP_NAME RUNTIME_B_NACOS_SERVICE_NAME RUNTIME_B_NACOS_CLUSTER_NAME +RUNTIME_B_NACOS_PORT_NAME RUNTIME_B_NACOS_ADVERTISED_IP RUNTIME_B_NACOS_ADVERTISED_PORT +RUNTIME_B_NACOS_WEIGHT RUNTIME_B_NACOS_HEARTBEAT_INTERVAL_MS +RUNTIME_B_NACOS_HEARTBEAT_TIMEOUT_MS +RUNTIME_B_NACOS_IP_DELETE_TIMEOUT_MS RUNTIME_B_NACOS_REQUEST_TIMEOUT_MS RUNTIME_B_NACOS_AUTH_MODE RUNTIME_B_NACOS_ACCESS_TOKEN @@ -50,12 +59,15 @@ the sample's JSON and SSE results so Stack acceptance can prove which replica handled an Invocation. It does not change the Agent ID, Release identity, Router credential audience, or nested-call authorization. -With `nacos` registration, Runtime B registers its ephemeral instance before -serving, sends one heartbeat per configured interval, and deregisters during -shutdown. A failed initial registration fails startup. A failed heartbeat -marks `/readyz` as `503` and stops serving; there is no retry or alternate -Nacos endpoint. `RUNTIME_B_NACOS_ACCESS_TOKEN` is required only for -`access_token` mode and is never logged. +With `nacos` registration, Runtime B uses Core's provider-neutral +`InstanceRegistrar` and `InstanceLease` contracts. The ready instance is bound +to one exact Agent Card/Release target before serving, its freshness values are +explicitly published, and shutdown closes the lease and deregisters it. A +failed initial registration fails startup. A terminal heartbeat failure closes +the lease, makes `/readyz` return `503`, and stops serving; there is no retry, +alternate Nacos endpoint, stale lease, or Release fallback. +`RUNTIME_B_NACOS_ACCESS_TOKEN` is required only for `access_token` mode and is +never logged. ## Test Runtime B diff --git a/runtime-b/cmd/runtime-b/main.go b/runtime-b/cmd/runtime-b/main.go index 3898101..e691fd0 100644 --- a/runtime-b/cmd/runtime-b/main.go +++ b/runtime-b/cmd/runtime-b/main.go @@ -35,7 +35,7 @@ func run() error { if err != nil { return err } - registrationConfig, err := runtimeb.LoadRegistrationConfig(os.LookupEnv, config.InstanceID) + registrationConfig, err := runtimeb.LoadRegistrationConfig(os.LookupEnv, config.AgentID, config.InstanceID) if err != nil { return err } diff --git a/runtime-b/config_test.go b/runtime-b/config_test.go index 31b61e1..88a8e8b 100644 --- a/runtime-b/config_test.go +++ b/runtime-b/config_test.go @@ -22,39 +22,48 @@ func validRuntimeBEnvironment() map[string]string { func validNacosRegistrationEnvironment() map[string]string { return map[string]string{ RegistrationModeEnvironment: RegistrationModeNacos, + AgentCardVersionEnvironment: "1.0.0", + ReleaseIDEnvironment: "rel_runtime_b_1", + CardDigestEnvironment: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + CanonicalEndpointEnvironment: "http://runtime-b:8092/", + AudienceEnvironment: "http://runtime-b:8092", NacosAPIOriginEnvironment: "http://nacos:8848/nacos", NacosNamespaceEnvironment: "public", NacosGroupEnvironment: "NEKIRO", NacosServiceEnvironment: "runtime-b", NacosClusterEnvironment: "DEFAULT", + NacosPortNameEnvironment: "a2a", NacosAdvertisedIPEnvironment: "172.28.0.12", NacosAdvertisedPortEnvironment: "8092", + NacosWeightEnvironment: "1", NacosHeartbeatIntervalEnvironment: "2000", + NacosHeartbeatTimeoutEnvironment: "5000", + NacosIPDeleteTimeoutEnvironment: "10000", NacosRequestTimeoutEnvironment: "1000", NacosAuthModeEnvironment: NacosAuthNone, } } func TestLoadRegistrationConfigRequiresExplicitModeAndNacosTuple(t *testing.T) { - disabled, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{RegistrationModeEnvironment: RegistrationModeDisabled}), "runtime-b-primary") + disabled, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{RegistrationModeEnvironment: RegistrationModeDisabled}), "runtime-b", "runtime-b-primary") if err != nil || disabled.Mode != RegistrationModeDisabled { t.Fatalf("disabled registration=%#v error=%v", disabled, err) } - configured, err := LoadRegistrationConfig(runtimeBLookup(validNacosRegistrationEnvironment()), "runtime-b-directory") + configured, err := LoadRegistrationConfig(runtimeBLookup(validNacosRegistrationEnvironment()), "runtime-b", "runtime-b-directory") if err != nil || configured.ServiceName != "runtime-b" || configured.AdvertisedIP != "172.28.0.12" || configured.AdvertisedPort != 8092 { t.Fatalf("Nacos registration=%#v error=%v", configured, err) } for name := range validNacosRegistrationEnvironment() { invalid := validNacosRegistrationEnvironment() delete(invalid, name) - if _, err := LoadRegistrationConfig(runtimeBLookup(invalid), "runtime-b-directory"); err == nil { + if _, err := LoadRegistrationConfig(runtimeBLookup(invalid), "runtime-b", "runtime-b-directory"); err == nil { t.Errorf("missing %s was accepted", name) } } - if _, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{}), "runtime-b-primary"); err == nil { + if _, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{}), "runtime-b", "runtime-b-primary"); err == nil { t.Fatal("missing registration mode was accepted") } - if _, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{RegistrationModeEnvironment: RegistrationModeDisabled, NacosServiceEnvironment: "unexpected"}), "runtime-b-primary"); err == nil { + if _, err := LoadRegistrationConfig(runtimeBLookup(map[string]string{RegistrationModeEnvironment: RegistrationModeDisabled, NacosServiceEnvironment: "unexpected"}), "runtime-b", "runtime-b-primary"); err == nil { t.Fatal("disabled registration accepted Nacos settings") } } @@ -62,17 +71,21 @@ func TestLoadRegistrationConfigRequiresExplicitModeAndNacosTuple(t *testing.T) { func TestLoadRegistrationConfigRejectsInvalidNacosSettings(t *testing.T) { for name, test := range map[string]struct{ key, value string }{ "origin": {NacosAPIOriginEnvironment, "http://nacos:8848"}, + "target": {CardDigestEnvironment, "not-a-digest"}, + "audience": {AudienceEnvironment, "http://another-runtime:8092"}, "namespace": {NacosNamespaceEnvironment, "not safe"}, "IP": {NacosAdvertisedIPEnvironment, "runtime-b"}, "port": {NacosAdvertisedPortEnvironment, "0"}, - "heartbeat": {NacosHeartbeatIntervalEnvironment, "99"}, + "heartbeat": {NacosHeartbeatIntervalEnvironment, "999"}, + "freshness": {NacosHeartbeatTimeoutEnvironment, "2000"}, + "expiry": {NacosIPDeleteTimeoutEnvironment, "5000"}, "timeout": {NacosRequestTimeoutEnvironment, "60001"}, "auth": {NacosAuthModeEnvironment, "implicit"}, } { t.Run(name, func(t *testing.T) { environment := validNacosRegistrationEnvironment() environment[test.key] = test.value - if _, err := LoadRegistrationConfig(runtimeBLookup(environment), "runtime-b-directory"); err == nil { + if _, err := LoadRegistrationConfig(runtimeBLookup(environment), "runtime-b", "runtime-b-directory"); err == nil { t.Fatalf("invalid %s=%q was accepted", test.key, test.value) } }) diff --git a/runtime-b/nacos_registration_test.go b/runtime-b/nacos_registration_test.go index 769d2f6..f8b06cd 100644 --- a/runtime-b/nacos_registration_test.go +++ b/runtime-b/nacos_registration_test.go @@ -14,6 +14,7 @@ import ( func TestNacosRegistrationOwnsRegisterHeartbeatAndDeregister(t *testing.T) { var mu sync.Mutex methods := make([]string, 0, 3) + heartbeat := make(chan struct{}, 1) server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { mu.Lock() methods = append(methods, request.Method+" "+request.URL.Path) @@ -22,37 +23,50 @@ func TestNacosRegistrationOwnsRegisterHeartbeatAndDeregister(t *testing.T) { t.Fatal(err) } query := request.Form - if query.Get("serviceName") != "runtime-b" || query.Get("groupName") != "NEKIRO" || query.Get("clusterName") != "DEFAULT" || query.Get("namespaceId") != "public" || query.Get("ip") != "127.0.0.1" || query.Get("port") != "8092" { + if query.Get("serviceName") != "NEKIRO@@runtime-b" || query.Get("groupName") != "NEKIRO" || query.Get("clusterName") != "DEFAULT" || query.Get("namespaceId") != "public" || query.Get("ip") != "127.0.0.1" || query.Get("port") != "8092" { t.Errorf("request query=%v", query) } if request.Method == http.MethodPost { var metadata map[string]string - if json.Unmarshal([]byte(query.Get("metadata")), &metadata) != nil || metadata["nekiro.instanceId"] != "runtime-b-directory" || query.Get("ephemeral") != "true" || query.Get("weight") != "1.0" { + if json.Unmarshal([]byte(query.Get("metadata")), &metadata) != nil || metadata["nekiro.instanceId"] != "runtime-b-directory" || metadata["preserved.heart.beat.interval"] != "1000" || metadata["preserved.heart.beat.timeout"] != "5000" || metadata["preserved.ip.delete.timeout"] != "10000" || query.Get("ephemeral") != "true" || query.Get("weight") != "1" { t.Errorf("registration metadata=%v query=%v", metadata, query) } } if request.Method == http.MethodPut { + select { + case heartbeat <- struct{}{}: + default: + } var beat struct { Service string `json:"serviceName"` Metadata map[string]string `json:"metadata"` Weight float64 `json:"weight"` } - if json.Unmarshal([]byte(query.Get("beat")), &beat) != nil || beat.Service != "NEKIRO@@runtime-b" || beat.Metadata["nekiro.instanceId"] != "runtime-b-directory" || beat.Weight != 1.0 || query.Get("ephemeral") != "true" { + if json.Unmarshal([]byte(query.Get("beat")), &beat) != nil || beat.Service != "NEKIRO@@runtime-b" || beat.Metadata["nekiro.instanceId"] != "runtime-b-directory" || beat.Metadata["preserved.heart.beat.timeout"] != "5000" || beat.Weight != 1.0 || query.Get("ephemeral") != "true" { t.Errorf("heartbeat=%v query=%v", beat, query) } } if request.Method == http.MethodPut { - _, _ = writer.Write([]byte(`{"clientBeatInterval":5000,"code":10200,"lightBeatEnabled":true}`)) + _, _ = writer.Write([]byte(`{"clientBeatInterval":1000,"code":10200,"lightBeatEnabled":true}`)) return } _, _ = writer.Write([]byte("ok")) })) t.Cleanup(server.Close) - registration := testRegistration(t, server, 10*time.Second) + registration := testRegistration(t, server, time.Second) if err := registration.Register(t.Context()); err != nil || !registration.Ready() { t.Fatalf("Register ready=%v error=%v", registration.Ready(), err) } - if err := registration.Heartbeat(t.Context()); err != nil { + runContext, cancelRun := context.WithCancel(t.Context()) + runErrors := make(chan error, 1) + go func() { runErrors <- registration.Run(runContext) }() + select { + case <-heartbeat: + case <-time.After(3 * time.Second): + t.Fatal("heartbeat was not sent") + } + cancelRun() + if err := <-runErrors; err != nil { t.Fatal(err) } if err := registration.Deregister(t.Context()); err != nil || registration.Ready() { @@ -82,7 +96,7 @@ func TestNacosHeartbeatFailureMakesRuntimeNotReadyWithoutRetry(t *testing.T) { _, _ = writer.Write([]byte("ok")) })) t.Cleanup(server.Close) - registration := testRegistration(t, server, 100*time.Millisecond) + registration := testRegistration(t, server, time.Second) if err := registration.Register(t.Context()); err != nil { t.Fatal(err) } @@ -118,6 +132,47 @@ func TestNacosRegistrationClassifiesCanceledAndUnavailableRequests(t *testing.T) } } +func TestNacosRegistrationRejectsInvalidLifecycleCalls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + _, _ = writer.Write([]byte("ok")) + })) + t.Cleanup(server.Close) + registration := testRegistration(t, server, time.Second) + if registration.Ready() { + t.Fatal("registration was ready before its initial publish") + } + if err := registration.Run(t.Context()); err == nil { + t.Fatal("lease observation started before registration") + } + if err := registration.Run(nil); err == nil { + t.Fatal("nil lease observation context was accepted") + } + if err := registration.Deregister(nil); err == nil { + t.Fatal("nil deregistration context was accepted") + } + if err := registration.Register(t.Context()); err != nil { + t.Fatal(err) + } + if err := registration.Register(t.Context()); err == nil { + t.Fatal("duplicate registration was accepted") + } + if err := registration.Deregister(t.Context()); err != nil { + t.Fatal(err) + } + if err := registration.Deregister(t.Context()); err != nil { + t.Fatalf("idempotent deregistration: %v", err) + } +} + +func TestNacosRegistrationRejectsInvalidExactRelease(t *testing.T) { + config := validRegistrationConfig("http://nacos.test/nacos", time.Second) + config.CardDigest = "not-a-digest" + registration, err := NewNacosRegistration(config, http.DefaultClient) + if err == nil || registration != nil { + t.Fatal("invalid exact Release target was accepted") + } +} + func testRegistration(t *testing.T, server *httptest.Server, interval time.Duration) *NacosRegistration { t.Helper() return testRegistrationWithURL(t, server.URL+"/nacos", server.Client(), interval) @@ -125,13 +180,22 @@ func testRegistration(t *testing.T, server *httptest.Server, interval time.Durat func testRegistrationWithURL(t *testing.T, origin string, client HTTPDoer, interval time.Duration) *NacosRegistration { t.Helper() - registration, err := NewNacosRegistration(RegistrationConfig{ - Mode: RegistrationModeNacos, APIOrigin: origin, NamespaceID: "public", GroupName: "NEKIRO", - ServiceName: "runtime-b", ClusterName: "DEFAULT", AdvertisedIP: "127.0.0.1", AdvertisedPort: 8092, - HeartbeatInterval: interval, RequestTimeout: time.Second, AuthMode: NacosAuthNone, InstanceID: "runtime-b-directory", - }, client) + registration, err := NewNacosRegistration(validRegistrationConfig(origin, interval), client) if err != nil { t.Fatal(err) } return registration } + +func validRegistrationConfig(origin string, interval time.Duration) RegistrationConfig { + return RegistrationConfig{ + Mode: RegistrationModeNacos, AgentID: "runtime-b", InstanceID: "runtime-b-directory", + AgentCardVersion: "1.0.0", ReleaseID: "rel_runtime_b_1", + CardDigest: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + CanonicalEndpoint: "http://runtime-b:8092/", Audience: "http://runtime-b:8092", + APIOrigin: origin, NamespaceID: "public", GroupName: "NEKIRO", ServiceName: "runtime-b", ClusterName: "DEFAULT", + PortName: "a2a", AdvertisedIP: "127.0.0.1", AdvertisedPort: 8092, Weight: 1, + HeartbeatInterval: interval, HeartbeatTimeout: 5 * time.Second, IPDeleteTimeout: 10 * time.Second, + RequestTimeout: time.Second, AuthMode: NacosAuthNone, + } +} diff --git a/runtime-b/registration_config.go b/runtime-b/registration_config.go index ec334b8..7202870 100644 --- a/runtime-b/registration_config.go +++ b/runtime-b/registration_config.go @@ -4,14 +4,23 @@ import "github.com/NeKiro-project/NeKiro-Samples/internal/nacosregistration" const ( RegistrationModeEnvironment = "RUNTIME_B_REGISTRATION_MODE" + AgentCardVersionEnvironment = "RUNTIME_B_AGENT_CARD_VERSION" + ReleaseIDEnvironment = "RUNTIME_B_RELEASE_ID" + CardDigestEnvironment = "RUNTIME_B_CARD_DIGEST" + CanonicalEndpointEnvironment = "RUNTIME_B_CANONICAL_ENDPOINT" + AudienceEnvironment = "RUNTIME_B_AUDIENCE" NacosAPIOriginEnvironment = "RUNTIME_B_NACOS_API_ORIGIN" NacosNamespaceEnvironment = "RUNTIME_B_NACOS_NAMESPACE_ID" NacosGroupEnvironment = "RUNTIME_B_NACOS_GROUP_NAME" NacosServiceEnvironment = "RUNTIME_B_NACOS_SERVICE_NAME" NacosClusterEnvironment = "RUNTIME_B_NACOS_CLUSTER_NAME" + NacosPortNameEnvironment = "RUNTIME_B_NACOS_PORT_NAME" NacosAdvertisedIPEnvironment = "RUNTIME_B_NACOS_ADVERTISED_IP" NacosAdvertisedPortEnvironment = "RUNTIME_B_NACOS_ADVERTISED_PORT" + NacosWeightEnvironment = "RUNTIME_B_NACOS_WEIGHT" NacosHeartbeatIntervalEnvironment = "RUNTIME_B_NACOS_HEARTBEAT_INTERVAL_MS" + NacosHeartbeatTimeoutEnvironment = "RUNTIME_B_NACOS_HEARTBEAT_TIMEOUT_MS" + NacosIPDeleteTimeoutEnvironment = "RUNTIME_B_NACOS_IP_DELETE_TIMEOUT_MS" NacosRequestTimeoutEnvironment = "RUNTIME_B_NACOS_REQUEST_TIMEOUT_MS" NacosAuthModeEnvironment = "RUNTIME_B_NACOS_AUTH_MODE" NacosAccessTokenEnvironment = "RUNTIME_B_NACOS_ACCESS_TOKEN" @@ -23,6 +32,6 @@ const ( type RegistrationConfig = nacosregistration.Config -func LoadRegistrationConfig(lookup func(string) (string, bool), instanceID string) (RegistrationConfig, error) { - return nacosregistration.Load(lookup, "RUNTIME_B", instanceID) +func LoadRegistrationConfig(lookup func(string) (string, bool), agentID, instanceID string) (RegistrationConfig, error) { + return nacosregistration.Load(lookup, "RUNTIME_B", agentID, instanceID) }