Skip to content

Commit 34b917a

Browse files
LionelJouinLionelJouin
committed
NSP Target Registry chained server
* New target registry server pattern allowing future feature * Current chain: watch handler -> store (sqlite + keepalive)
1 parent 61c9766 commit 34b917a

File tree

7 files changed

+356
-68
lines changed

7 files changed

+356
-68
lines changed

‎cmd/nsp/main.go

+39-18
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,9 @@ import (
3232
"github.com/nordix/meridio/pkg/configuration/registry"
3333
"github.com/nordix/meridio/pkg/health"
3434
"github.com/nordix/meridio/pkg/nsp"
35+
nspRegistry "github.com/nordix/meridio/pkg/nsp/registry"
36+
"github.com/nordix/meridio/pkg/nsp/watchhandler"
37+
"github.com/pkg/errors"
3538

3639
keepAliveRegistry "github.com/nordix/meridio/pkg/nsp/registry/keepalive"
3740
sqliteRegistry "github.com/nordix/meridio/pkg/nsp/registry/sqlite"
@@ -97,31 +100,18 @@ func main() {
97100
ctx = health.CreateChecker(ctx)
98101

99102
// configuration
100-
configurationEventChan := make(chan *registry.ConfigurationEvent, 10)
101-
configurationRegistry := registry.New(configurationEventChan)
102-
configurationMonitor, err := monitor.New(config.ConfigMapName, config.Namespace, configurationRegistry)
103+
configurationManagerServer, err := CreateConfigurationManagerServer(ctx, &config)
103104
if err != nil {
104-
logrus.Fatalf("Unable to start configuration monitor: %v", err)
105+
logrus.Fatalf("CreateConfigurationManagerServer err: %v", err)
105106
}
106-
go configurationMonitor.Start(context.Background())
107-
watcherNotifier := manager.NewWatcherNotifier(configurationRegistry, configurationEventChan)
108-
go watcherNotifier.Start(context.Background())
109-
configurationManagerServer := manager.NewServer(watcherNotifier)
110107

111108
// target registry
112-
sqlr, err := sqliteRegistry.New(config.Datasource)
109+
targetRegistryServer, err := CreateTargetRegistryServer(ctx, &config)
113110
if err != nil {
114-
logrus.Fatalf("Unable create sqlite registry: %v", err)
111+
logrus.Fatalf("CreateTargetRegistryServer err: %v", err)
115112
}
116-
keepAliveRegistry, err := keepAliveRegistry.New(
117-
keepAliveRegistry.WithRegistry(sqlr),
118-
keepAliveRegistry.WithTimeout(config.EntryTimeout),
119-
)
120-
if err != nil {
121-
logrus.Fatalf("Unable create keepalive registry: %v", err)
122-
}
123-
targetRegistryServer := nsp.NewServer(keepAliveRegistry)
124113

114+
// Create Server
125115
server := grpc.NewServer(grpc.Creds(
126116
credentials.GetServer(context.Background()),
127117
))
@@ -141,3 +131,34 @@ func main() {
141131

142132
<-ctx.Done()
143133
}
134+
135+
func CreateTargetRegistryServer(ctx context.Context, config *Config) (nspAPI.TargetRegistryServer, error) {
136+
sqlr, err := sqliteRegistry.New(config.Datasource)
137+
if err != nil {
138+
return nil, errors.Wrap(err, "Unable create sqlite registry")
139+
}
140+
keepAliveRegistry, err := keepAliveRegistry.New(
141+
keepAliveRegistry.WithRegistry(sqlr),
142+
keepAliveRegistry.WithTimeout(config.EntryTimeout),
143+
)
144+
if err != nil {
145+
return nil, errors.Wrap(err, "Unable create keepalive registry")
146+
}
147+
return nsp.NewServer(
148+
nspRegistry.NewServer(keepAliveRegistry),
149+
watchhandler.NewServer(keepAliveRegistry),
150+
), nil
151+
}
152+
153+
func CreateConfigurationManagerServer(ctx context.Context, config *Config) (nspAPI.ConfigurationManagerServer, error) {
154+
configurationEventChan := make(chan *registry.ConfigurationEvent, 10)
155+
configurationRegistry := registry.New(configurationEventChan)
156+
configurationMonitor, err := monitor.New(config.ConfigMapName, config.Namespace, configurationRegistry)
157+
if err != nil {
158+
return nil, errors.Wrap(err, "Unable to start configuration monitor")
159+
}
160+
go configurationMonitor.Start(context.Background())
161+
watcherNotifier := manager.NewWatcherNotifier(configurationRegistry, configurationEventChan)
162+
go watcherNotifier.Start(context.Background())
163+
return manager.NewServer(watcherNotifier), nil
164+
}

‎pkg/nsp/next/builder.go

+33
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
Copyright (c) 2022 Nordix Foundation
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package next
18+
19+
// BuildNextTargetRegistryChain chains the target registry servers together.
20+
// Each NextTargetRegistryServer must have a non nil NextTargetRegistryServerImpl.
21+
// If the list of nextTargetRegistryServers is nil or empty, a nil value will be returned.
22+
func BuildNextTargetRegistryChain(nextTargetRegistryServers ...NextTargetRegistryServer) NextTargetRegistryServer {
23+
if len(nextTargetRegistryServers) <= 0 {
24+
return nil
25+
}
26+
for i, ntrs := range nextTargetRegistryServers {
27+
if i >= (len(nextTargetRegistryServers) - 1) {
28+
break
29+
}
30+
ntrs.setNext(nextTargetRegistryServers[i+1])
31+
}
32+
return nextTargetRegistryServers[0]
33+
}

‎pkg/nsp/next/builder_test.go

+83
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
/*
2+
Copyright (c) 2022 Nordix Foundation
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package next_test
18+
19+
import (
20+
"reflect"
21+
"testing"
22+
23+
"github.com/nordix/meridio/pkg/nsp/next"
24+
)
25+
26+
type fakeNextTargetRegistryServerImpl struct {
27+
*next.NextTargetRegistryServerImpl
28+
name string
29+
}
30+
31+
func TestBuildNextTargetRegistryChain(t *testing.T) {
32+
fakeA := &fakeNextTargetRegistryServerImpl{
33+
NextTargetRegistryServerImpl: &next.NextTargetRegistryServerImpl{},
34+
name: "a",
35+
}
36+
type args struct {
37+
nextTargetRegistryServers []next.NextTargetRegistryServer
38+
}
39+
tests := []struct {
40+
name string
41+
args args
42+
want next.NextTargetRegistryServer
43+
}{
44+
{
45+
name: "empty",
46+
args: args{},
47+
want: nil,
48+
},
49+
{
50+
name: "one",
51+
args: args{
52+
[]next.NextTargetRegistryServer{
53+
fakeA,
54+
},
55+
},
56+
want: fakeA,
57+
},
58+
{
59+
name: "multiple",
60+
args: args{
61+
[]next.NextTargetRegistryServer{
62+
fakeA,
63+
&fakeNextTargetRegistryServerImpl{
64+
NextTargetRegistryServerImpl: &next.NextTargetRegistryServerImpl{},
65+
name: "b",
66+
},
67+
&fakeNextTargetRegistryServerImpl{
68+
NextTargetRegistryServerImpl: &next.NextTargetRegistryServerImpl{},
69+
name: "c",
70+
},
71+
},
72+
},
73+
want: fakeA,
74+
},
75+
}
76+
for _, tt := range tests {
77+
t.Run(tt.name, func(t *testing.T) {
78+
if got := next.BuildNextTargetRegistryChain(tt.args.nextTargetRegistryServers...); !reflect.DeepEqual(got, tt.want) {
79+
t.Errorf("BuildNextTargetRegistryChain() = %v, want %v", got, tt.want)
80+
}
81+
})
82+
}
83+
}

‎pkg/nsp/next/next.go

+67
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
/*
2+
Copyright (c) 2021 Nordix Foundation
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package next
18+
19+
import (
20+
"context"
21+
22+
"github.com/golang/protobuf/ptypes/empty"
23+
nspAPI "github.com/nordix/meridio/api/nsp/v1"
24+
)
25+
26+
// NextTargetRegistryServer is the interface representing TargetRegistryServer with the
27+
// support of the chaining feature.
28+
type NextTargetRegistryServer interface {
29+
nspAPI.TargetRegistryServer
30+
setNext(NextTargetRegistryServer)
31+
}
32+
33+
type NextTargetRegistryServerImpl struct {
34+
nspAPI.UnimplementedTargetRegistryServer
35+
next NextTargetRegistryServer
36+
}
37+
38+
// Register will call the Register function of the next chain element
39+
// If the next element is nil, then &empty.Empty{}, nil will be returned.
40+
func (ntrsi *NextTargetRegistryServerImpl) Register(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
41+
if ntrsi.next == nil {
42+
return &empty.Empty{}, nil
43+
}
44+
return ntrsi.next.Register(ctx, target)
45+
}
46+
47+
// Unregister will call the Unregister function of the next chain element
48+
// If the next element is nil, then &empty.Empty{}, nil will be returned.
49+
func (ntrsi *NextTargetRegistryServerImpl) Unregister(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
50+
if ntrsi.next == nil {
51+
return &empty.Empty{}, nil
52+
}
53+
return ntrsi.next.Unregister(ctx, target)
54+
}
55+
56+
// Watch will call the Watch function of the next chain element
57+
// If the next element is nil, then nil will be returned.
58+
func (ntrsi *NextTargetRegistryServerImpl) Watch(t *nspAPI.Target, watcher nspAPI.TargetRegistry_WatchServer) error {
59+
if ntrsi.next == nil {
60+
return nil
61+
}
62+
return ntrsi.next.Watch(t, watcher)
63+
}
64+
65+
func (ntrsi *NextTargetRegistryServerImpl) setNext(ntrs NextTargetRegistryServer) {
66+
ntrsi.next = ntrs
67+
}

‎pkg/nsp/registry/server.go

+59
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
/*
2+
Copyright (c) 2021 Nordix Foundation
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package registry
18+
19+
import (
20+
"context"
21+
22+
"github.com/golang/protobuf/ptypes/empty"
23+
nspAPI "github.com/nordix/meridio/api/nsp/v1"
24+
"github.com/nordix/meridio/pkg/nsp/next"
25+
"github.com/nordix/meridio/pkg/nsp/types"
26+
)
27+
28+
type registry struct {
29+
TargetRegistry types.TargetRegistry
30+
*next.NextTargetRegistryServerImpl
31+
}
32+
33+
// NewServer provides an implementation of TargetRegistryServer with the
34+
// support of the chaining feature. This implementation handles Register
35+
// and Unregister calls by adding or removing data into a storage (e.g
36+
// memory or sqlite)
37+
func NewServer(targetRegistry types.TargetRegistry) *registry {
38+
r := &registry{
39+
TargetRegistry: targetRegistry,
40+
NextTargetRegistryServerImpl: &next.NextTargetRegistryServerImpl{},
41+
}
42+
return r
43+
}
44+
45+
func (r *registry) Register(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
46+
err := r.TargetRegistry.Set(ctx, target)
47+
if err != nil {
48+
return &empty.Empty{}, err
49+
}
50+
return r.NextTargetRegistryServerImpl.Register(ctx, target)
51+
}
52+
53+
func (r *registry) Unregister(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
54+
err := r.TargetRegistry.Remove(ctx, target)
55+
if err != nil {
56+
return &empty.Empty{}, err
57+
}
58+
return r.NextTargetRegistryServerImpl.Unregister(ctx, target)
59+
}

‎pkg/nsp/server.go

+3-50
Original file line numberDiff line numberDiff line change
@@ -17,58 +17,11 @@ limitations under the License.
1717
package nsp
1818

1919
import (
20-
"context"
21-
22-
"github.com/golang/protobuf/ptypes/empty"
2320
nspAPI "github.com/nordix/meridio/api/nsp/v1"
24-
"github.com/nordix/meridio/pkg/nsp/types"
25-
"github.com/sirupsen/logrus"
21+
"github.com/nordix/meridio/pkg/nsp/next"
2622
)
2723

28-
type Server struct {
29-
nspAPI.UnimplementedTargetRegistryServer
30-
TargetRegistry types.TargetRegistry
31-
}
32-
3324
// NewServer -
34-
func NewServer(targetRegistry types.TargetRegistry) nspAPI.TargetRegistryServer {
35-
networkServicePlateformService := &Server{
36-
TargetRegistry: targetRegistry,
37-
}
38-
39-
return networkServicePlateformService
40-
}
41-
42-
func (s *Server) Register(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
43-
return &empty.Empty{}, s.TargetRegistry.Set(ctx, target)
44-
}
45-
46-
func (s *Server) Unregister(ctx context.Context, target *nspAPI.Target) (*empty.Empty, error) {
47-
return &empty.Empty{}, s.TargetRegistry.Remove(ctx, target)
48-
}
49-
50-
func (s *Server) Watch(t *nspAPI.Target, watcher nspAPI.TargetRegistry_WatchServer) error {
51-
targetWatcher, err := s.TargetRegistry.Watch(context.TODO(), t)
52-
if err != nil {
53-
return err
54-
}
55-
s.watcher(watcher, targetWatcher.ResultChan())
56-
targetWatcher.Stop()
57-
return nil
58-
}
59-
60-
func (s *Server) watcher(watcher nspAPI.TargetRegistry_WatchServer, ch <-chan []*nspAPI.Target) {
61-
for {
62-
select {
63-
case event := <-ch:
64-
err := watcher.Send(&nspAPI.TargetResponse{
65-
Targets: event,
66-
})
67-
if err != nil {
68-
logrus.Errorf("err sending TrenchResponse: %v", err)
69-
}
70-
case <-watcher.Context().Done():
71-
return
72-
}
73-
}
25+
func NewServer(nextTargetRegistryServers ...next.NextTargetRegistryServer) nspAPI.TargetRegistryServer {
26+
return next.BuildNextTargetRegistryChain(nextTargetRegistryServers...)
7427
}

0 commit comments

Comments
 (0)