Skip to content

Commit 9a84dad

Browse files
tinyzimmertinyzimmer
committed
tidy: reorganize plugindb and pull in API support for writes
1 parent 697179c commit 9a84dad

File tree

7 files changed

+178
-132
lines changed

7 files changed

+178
-132
lines changed

‎go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ require (
4848
github.com/spf13/cobra v1.7.0
4949
github.com/spf13/pflag v1.0.5
5050
github.com/vishvananda/netlink v1.2.1-beta.2
51-
github.com/webmeshproj/api v0.10.1
51+
github.com/webmeshproj/api v0.10.2
5252
golang.org/x/crypto v0.14.0
5353
golang.org/x/exp v0.0.0-20231006140011-7918f672742d
5454
golang.org/x/sync v0.4.0

‎go.sum

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1572,8 +1572,8 @@ github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1Y
15721572
github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
15731573
github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSDJfjId/PEGEShv6ugrt4kYsC5UIDaQ=
15741574
github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=
1575-
github.com/webmeshproj/api v0.10.1 h1:3e2/C2KcYL2JEBrmBS/KgjxxEkVxkwKWxU/g6w645x8=
1576-
github.com/webmeshproj/api v0.10.1/go.mod h1:lFGd4eyG9wr0mQiAJxQDBJ6VwEe5EC0AXKZjTv6ZA3U=
1575+
github.com/webmeshproj/api v0.10.2 h1:3LQPB9A1s1o3NSQX2JlAzY1stncIcrOWa4q/QjaaU6I=
1576+
github.com/webmeshproj/api v0.10.2/go.mod h1:lFGd4eyG9wr0mQiAJxQDBJ6VwEe5EC0AXKZjTv6ZA3U=
15771577
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1 h1:EKhdznlJHPMoKr0XTrX+IlJs1LH3lyx2nfr1dOlZ79k=
15781578
github.com/whyrusleeping/go-keyspace v0.0.0-20160322163242-5b898ac5add1/go.mod h1:8UvriyWtv5Q5EOgjHaSseUEdkQfvwFv1I/In/O2M9gc=
15791579
github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=

‎pkg/plugins/plugindb/plugindb_client.go renamed to ‎pkg/plugins/plugindb/db_client.go

Lines changed: 0 additions & 96 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,6 @@ See the License for the specific language governing permissions and
1414
limitations under the License.
1515
*/
1616

17-
// Package plugindb contains an interface for performing storage queries
18-
// over the storage APIs.
1917
package plugindb
2018

2119
import (
@@ -24,7 +22,6 @@ import (
2422
"strconv"
2523
"strings"
2624
"sync"
27-
"time"
2825

2926
"github.com/dominikbraun/graph"
3027
v1 "github.com/webmeshproj/api/go/v1"
@@ -52,99 +49,6 @@ func OpenDB(srv QueryServer) storage.MeshDB {
5249
return meshdb.New(&PluginDataStore{QueryServer: srv})
5350
}
5451

55-
// OpenKeyVal opens a new key-value store connection to a plugin query stream.
56-
func OpenKeyVal(srv QueryServer) storage.MeshStorage {
57-
return &PluginMeshStorage{QueryServer: srv}
58-
}
59-
60-
// PluginMeshStorage implements a mesh key-value store over a plugin query stream.
61-
type PluginMeshStorage struct {
62-
QueryServer
63-
mu sync.Mutex
64-
}
65-
66-
// GetValue returns the value of a key.
67-
func (p *PluginMeshStorage) GetValue(ctx context.Context, key []byte) ([]byte, error) {
68-
p.mu.Lock()
69-
defer p.mu.Unlock()
70-
if !types.IsValidPathID(string(key)) {
71-
return nil, errors.ErrInvalidKey
72-
}
73-
err := p.Send(&v1.QueryRequest{
74-
Command: v1.QueryRequest_GET,
75-
Type: v1.QueryRequest_VALUE,
76-
Query: types.NewQueryFilters().WithID(string(key)).Encode(),
77-
})
78-
if err != nil {
79-
return nil, err
80-
}
81-
resp, err := p.Recv()
82-
if err != nil {
83-
return nil, err
84-
}
85-
if resp.GetError() != "" {
86-
if strings.Contains(err.Error(), "not found") {
87-
return nil, errors.ErrKeyNotFound
88-
}
89-
return nil, fmt.Errorf(resp.GetError())
90-
}
91-
if len(resp.GetItems()) == 0 {
92-
return nil, errors.ErrKeyNotFound
93-
}
94-
return resp.GetItems()[0], nil
95-
}
96-
97-
func (p *PluginMeshStorage) PutValue(ctx context.Context, key, value []byte, ttl time.Duration) error {
98-
return errors.ErrNotStorageNode
99-
}
100-
101-
func (p *PluginMeshStorage) Delete(ctx context.Context, key []byte) error {
102-
return errors.ErrNotStorageNode
103-
}
104-
105-
func (p *PluginMeshStorage) ListKeys(ctx context.Context, prefix []byte) ([][]byte, error) {
106-
p.mu.Lock()
107-
defer p.mu.Unlock()
108-
err := p.Send(&v1.QueryRequest{
109-
Command: v1.QueryRequest_LIST,
110-
Type: v1.QueryRequest_KEYS,
111-
Query: types.NewQueryFilters().WithID(string(prefix)).Encode(),
112-
})
113-
if err != nil {
114-
return nil, err
115-
}
116-
resp, err := p.Recv()
117-
if err != nil {
118-
return nil, err
119-
}
120-
return resp.GetItems(), nil
121-
}
122-
123-
func (p *PluginMeshStorage) IterPrefix(ctx context.Context, prefix []byte, fn storage.PrefixIterator) error {
124-
keys, err := p.ListKeys(ctx, prefix)
125-
if err != nil {
126-
return err
127-
}
128-
for _, key := range keys {
129-
value, err := p.GetValue(ctx, key)
130-
if err != nil {
131-
return err
132-
}
133-
if err := fn(key, value); err != nil {
134-
return err
135-
}
136-
}
137-
return nil
138-
}
139-
140-
func (p *PluginMeshStorage) Subscribe(ctx context.Context, prefix []byte, fn storage.KVSubscribeFunc) (context.CancelFunc, error) {
141-
return func() {}, errors.ErrNotStorageNode
142-
}
143-
144-
func (p *PluginMeshStorage) Close() error {
145-
return nil
146-
}
147-
14852
// PluginDataStore implements a mesh data store over a plugin query stream.
14953
type PluginDataStore struct {
15054
QueryServer

‎pkg/plugins/plugindb/plugindb_serve.go renamed to ‎pkg/plugins/plugindb/db_server.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,6 @@ See the License for the specific language governing permissions and
1414
limitations under the License.
1515
*/
1616

17-
// Package plugindb contains an interface for performing storage queries
18-
// over the storage APIs.
1917
package plugindb
2018

2119
import (

‎pkg/plugins/plugindb/kv_client.go

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
/*
2+
Copyright 2023 Avi Zimmerman <[email protected]>
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 plugindb
18+
19+
import (
20+
"fmt"
21+
"strings"
22+
"sync"
23+
"time"
24+
25+
v1 "github.com/webmeshproj/api/go/v1"
26+
27+
"github.com/webmeshproj/webmesh/pkg/context"
28+
"github.com/webmeshproj/webmesh/pkg/storage"
29+
"github.com/webmeshproj/webmesh/pkg/storage/errors"
30+
"github.com/webmeshproj/webmesh/pkg/storage/types"
31+
)
32+
33+
// OpenKeyVal opens a new key-value store connection to a plugin query stream.
34+
func OpenKeyVal(srv QueryServer) storage.MeshStorage {
35+
return &PluginMeshStorage{QueryServer: srv}
36+
}
37+
38+
// PluginMeshStorage implements a mesh key-value store over a plugin query stream.
39+
type PluginMeshStorage struct {
40+
QueryServer
41+
mu sync.Mutex
42+
}
43+
44+
// GetValue returns the value of a key.
45+
func (p *PluginMeshStorage) GetValue(ctx context.Context, key []byte) ([]byte, error) {
46+
p.mu.Lock()
47+
defer p.mu.Unlock()
48+
if !types.IsValidPathID(string(key)) {
49+
return nil, errors.ErrInvalidKey
50+
}
51+
err := p.Send(&v1.QueryRequest{
52+
Command: v1.QueryRequest_GET,
53+
Type: v1.QueryRequest_VALUE,
54+
Query: types.NewQueryFilters().WithID(string(key)).Encode(),
55+
})
56+
if err != nil {
57+
return nil, err
58+
}
59+
resp, err := p.Recv()
60+
if err != nil {
61+
return nil, err
62+
}
63+
if resp.GetError() != "" {
64+
if strings.Contains(err.Error(), "not found") {
65+
return nil, errors.ErrKeyNotFound
66+
}
67+
return nil, fmt.Errorf(resp.GetError())
68+
}
69+
if len(resp.GetItems()) == 0 {
70+
return nil, errors.ErrKeyNotFound
71+
}
72+
return resp.GetItems()[0], nil
73+
}
74+
75+
func (p *PluginMeshStorage) PutValue(ctx context.Context, key, value []byte, ttl time.Duration) error {
76+
return errors.ErrNotStorageNode
77+
}
78+
79+
func (p *PluginMeshStorage) Delete(ctx context.Context, key []byte) error {
80+
return errors.ErrNotStorageNode
81+
}
82+
83+
func (p *PluginMeshStorage) ListKeys(ctx context.Context, prefix []byte) ([][]byte, error) {
84+
p.mu.Lock()
85+
defer p.mu.Unlock()
86+
err := p.Send(&v1.QueryRequest{
87+
Command: v1.QueryRequest_LIST,
88+
Type: v1.QueryRequest_KEYS,
89+
Query: types.NewQueryFilters().WithID(string(prefix)).Encode(),
90+
})
91+
if err != nil {
92+
return nil, err
93+
}
94+
resp, err := p.Recv()
95+
if err != nil {
96+
return nil, err
97+
}
98+
return resp.GetItems(), nil
99+
}
100+
101+
func (p *PluginMeshStorage) IterPrefix(ctx context.Context, prefix []byte, fn storage.PrefixIterator) error {
102+
keys, err := p.ListKeys(ctx, prefix)
103+
if err != nil {
104+
return err
105+
}
106+
for _, key := range keys {
107+
value, err := p.GetValue(ctx, key)
108+
if err != nil {
109+
return err
110+
}
111+
if err := fn(key, value); err != nil {
112+
return err
113+
}
114+
}
115+
return nil
116+
}
117+
118+
func (p *PluginMeshStorage) Subscribe(ctx context.Context, prefix []byte, fn storage.KVSubscribeFunc) (context.CancelFunc, error) {
119+
return func() {}, errors.ErrNotStorageNode
120+
}
121+
122+
func (p *PluginMeshStorage) Close() error {
123+
return nil
124+
}

‎pkg/plugins/plugindb/plugindb.go

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
/*
2+
Copyright 2023 Avi Zimmerman <[email protected]>
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 plugindb contains an interface for performing storage queries
18+
// over the storage APIs.
19+
package plugindb

‎pkg/storage/types/storage_queries.go

Lines changed: 32 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ type StorageQuery struct {
3333

3434
// ParseStorageQuery parses a storage query.
3535
func ParseStorageQuery(query *v1.QueryRequest) (StorageQuery, error) {
36-
filters := parseFilters(query)
36+
filters := ParseQueryFilters(query)
3737
if query.GetCommand() == v1.QueryRequest_GET {
3838
// The filter should always contain an ID or pub key
3939
// unless its for raw network or rbac state.
@@ -113,7 +113,7 @@ const (
113113
// IsValid returns true if the filter type is valid.
114114
func (f FilterType) IsValid() bool {
115115
switch f {
116-
case FilterTypeID, FilterTypePubKey:
116+
case FilterTypeID, FilterTypePubKey, FilterTypeSourceID, FilterTypeTargetID, FilterTypeNodeID, FilterTypeCIDR:
117117
return true
118118
default:
119119
return false
@@ -123,11 +123,41 @@ func (f FilterType) IsValid() bool {
123123
// QueryFilters is a list of parsed filters for a storage query.
124124
type QueryFilters []QueryFilter
125125

126+
// QueryFilter is a parsed filter for a storage query.
127+
type QueryFilter struct {
128+
// The type of filter.
129+
Type FilterType
130+
// Value is the value of the filter.
131+
Value string
132+
}
133+
126134
// NewQueryFilters returns a new list of query filters.
127135
func NewQueryFilters(filters ...QueryFilter) QueryFilters {
128136
return filters
129137
}
130138

139+
// ParseQueryFilters parses the query filters from a query request.
140+
func ParseQueryFilters(req *v1.QueryRequest) QueryFilters {
141+
query := req.GetQuery()
142+
fields := strings.Split(query, ",")
143+
filters := make(QueryFilters, 0, len(fields))
144+
for _, field := range fields {
145+
parts := strings.Split(field, "=")
146+
if len(parts) != 2 {
147+
continue
148+
}
149+
ftype := FilterType(parts[0])
150+
if !ftype.IsValid() {
151+
continue
152+
}
153+
filters = append(filters, QueryFilter{
154+
Type: ftype,
155+
Value: parts[1],
156+
})
157+
}
158+
return filters
159+
}
160+
131161
// Encode encodes the query filters into a string.
132162
func (q QueryFilters) Encode() string {
133163
var sb strings.Builder
@@ -250,32 +280,3 @@ func (q QueryFilters) GetByType(ftype FilterType) (QueryFilter, bool) {
250280
}
251281
return QueryFilter{}, false
252282
}
253-
254-
// QueryFilter is a parsed filter for a storage query.
255-
type QueryFilter struct {
256-
// The type of filter.
257-
Type FilterType
258-
// Value is the value of the filter.
259-
Value string
260-
}
261-
262-
func parseFilters(req *v1.QueryRequest) QueryFilters {
263-
query := req.GetQuery()
264-
fields := strings.Split(query, ",")
265-
filters := make(QueryFilters, 0, len(fields))
266-
for _, field := range fields {
267-
parts := strings.Split(field, "=")
268-
if len(parts) != 2 {
269-
continue
270-
}
271-
ftype := FilterType(parts[0])
272-
if !ftype.IsValid() {
273-
continue
274-
}
275-
filters = append(filters, QueryFilter{
276-
Type: ftype,
277-
Value: parts[1],
278-
})
279-
}
280-
return filters
281-
}

0 commit comments

Comments
 (0)