Skip to content

Commit 1a9f4cb

Browse files
calebdoxseycalebdoxsey
authored
cmd: add additional databroker options (#1229)
## Summary Add some additional cli options: - `grpc-addr`: sets the gRPC address, defaults to `:5443` - `services`: sets the services to run, defaults to `all` - `databroker-cluster-node-id`: sets the databroker cluster node id, defaults to unset - `databroker-cluster-nodes`: a JSON array of cluster nodes, defaults to nil - `databroker-raft-bind-address`: the databroker raft bind address, defaults to unset - `databroker-service-urls`: the databroker service urls, defaults to localhost, when set the outbound port is used to make the connection to the databroker And new CRD options: - `databroker/clusterLeaderId`: sets the databroker cluster leader, defaults to unset These additional options should be backwards compatible and not change any existing installation. ## Related issues - [ENG-2882](https://linear.app/pomerium/issue/ENG-2882/ingress-controller-add-clustered-databroker-configuration) ## Checklist - [x] reference any related issues - [ ] updated docs - [x] updated unit tests - [ ] updated UPGRADING.md - [x] add appropriate tag (`improvement` / `bug` / etc) - [x] ready for review
1 parent 988257f commit 1a9f4cb

11 files changed

Lines changed: 288 additions & 47 deletions

File tree

‎apis/ingress/v1/pomerium_types.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -353,6 +353,9 @@ type PomeriumSpec struct {
353353
// CircuitBreakerThresholds sets the circuit breaker thresholds settings.
354354
CircuitBreakerThresholds *CircuitBreakerThresholds `json:"circuitBreakerThresholds,omitempty"`
355355

356+
// DataBroker sets the databroker settings.
357+
DataBroker *DataBroker `json:"dataBroker,omitempty"`
358+
356359
// DNS sets the dns settings.
357360
DNS *DNS `json:"dns,omitempty"`
358361

@@ -465,6 +468,13 @@ type CircuitBreakerThresholds struct {
465468
MaxConnectionPools *uint32 `json:"maxConnectionPools"`
466469
}
467470

471+
// DataBroker are the databroker settings.
472+
type DataBroker struct {
473+
// ClusterLeaderID defines the cluster leader in a clustered databroker.
474+
// +kubebuilder:validation:Optional
475+
ClusterLeaderID *string `json:"clusterLeaderId,omitempty"`
476+
}
477+
468478
// DNS are the dns settings.
469479
type DNS struct {
470480
// LookupFamily is the DNS IP address resolution policy.

‎apis/ingress/v1/zz_generated.deepcopy.go‎

Lines changed: 25 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎cmd/all_in_one.go‎

Lines changed: 23 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -2,18 +2,16 @@ package cmd
22

33
import (
44
"context"
5-
"encoding/base64"
65
"fmt"
7-
"net"
86
"net/url"
97
"os"
8+
"strings"
109
"time"
1110

1211
validate "github.com/go-playground/validator/v10"
1312
"github.com/spf13/cobra"
1413
"github.com/volatiletech/null/v9"
1514
"golang.org/x/sync/errgroup"
16-
"google.golang.org/grpc"
1715
"google.golang.org/protobuf/proto"
1816
"k8s.io/apimachinery/pkg/types"
1917
"k8s.io/utils/ptr"
@@ -22,13 +20,6 @@ import (
2220
"sigs.k8s.io/controller-runtime/pkg/log"
2321
metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
2422

25-
"github.com/pomerium/pomerium/config"
26-
"github.com/pomerium/pomerium/pkg/grpc/databroker"
27-
"github.com/pomerium/pomerium/pkg/grpcutil"
28-
"github.com/pomerium/pomerium/pkg/health"
29-
"github.com/pomerium/pomerium/pkg/netutil"
30-
"github.com/pomerium/pomerium/pkg/telemetry/trace"
31-
3223
"github.com/pomerium/ingress-controller/controllers"
3324
"github.com/pomerium/ingress-controller/controllers/gateway"
3425
"github.com/pomerium/ingress-controller/controllers/ingress"
@@ -37,6 +28,11 @@ import (
3728
pomerium_ctrl "github.com/pomerium/ingress-controller/pomerium/ctrl"
3829
"github.com/pomerium/ingress-controller/util"
3930
health_ctrl "github.com/pomerium/ingress-controller/util/health"
31+
"github.com/pomerium/pomerium/config"
32+
"github.com/pomerium/pomerium/pkg/grpc/databroker"
33+
"github.com/pomerium/pomerium/pkg/health"
34+
"github.com/pomerium/pomerium/pkg/netutil"
35+
"github.com/pomerium/pomerium/pkg/telemetry/trace"
4036
)
4137

4238
type allCmdOptions struct {
@@ -47,14 +43,18 @@ type allCmdOptions struct {
4743
debugEnvoy bool
4844
adminBindAddr string
4945
configControllerShutdownTimeout time.Duration
50-
// metricsBindAddress must be externally accessible host:port
51-
metricsBindAddress string `validate:"required,hostname_port"`
5246
// healthProbeBindAddress must be externally accessible host:port
5347
healthProbeBindAddress string
54-
serverAddr string `validate:"required,hostname_port"`
55-
sshAddr string
56-
httpRedirectAddr string `validate:"required,hostname_port"`
57-
deriveTLS string `validate:"required,hostname"`
48+
// metricsBindAddress must be externally accessible host:port
49+
metricsBindAddress string `validate:"required,hostname_port"`
50+
serverAddr string `validate:"required,hostname_port"`
51+
sshAddr string
52+
httpRedirectAddr string `validate:"required,hostname_port"`
53+
deriveTLS string `validate:"required,hostname"`
54+
grpcAddr string `validate:"required,hostname_port"`
55+
services []string `validate:"dive,oneof=all authenticate authorize databroker proxy"`
56+
57+
DataBrokerOptions dataBrokerOptions
5858
}
5959

6060
type allCmdParam struct {
@@ -129,6 +129,8 @@ func (s *allCmd) setupFlags() error {
129129
flags.StringVar(&s.httpRedirectAddr, "http-redirect-addr", ":8080", "the address HTTP redirect would bind to")
130130
flags.StringVar(&s.deriveTLS, "databroker-auto-tls", "", "enable auto TLS and generate server certificate for the domain")
131131
flags.DurationVar(&s.configControllerShutdownTimeout, configControllerShutdown, time.Second*30, "timeout waiting for graceful config controller shutdown")
132+
flags.StringVar(&s.grpcAddr, "grpc-addr", ":5443", "the address the gRPC server would bind to")
133+
flags.StringSliceVar(&s.services, "services", []string{"all"}, "the pomerium services to run")
132134

133135
for _, flag := range hidden {
134136
if err := s.PersistentFlags().MarkHidden(flag); err != nil {
@@ -137,6 +139,7 @@ func (s *allCmd) setupFlags() error {
137139
}
138140

139141
s.ingressControllerOpts.setupFlags(flags)
142+
s.DataBrokerOptions.setupFlags(flags)
140143
return viperWalk(flags)
141144
}
142145

@@ -234,7 +237,9 @@ func (s *allCmdParam) run(ctx context.Context) error {
234237
func (s *allCmdParam) makeBootstrapConfig(opt allCmdOptions) error {
235238
s.cfg.Options = config.NewDefaultOptions()
236239

240+
s.cfg.Options.Services = strings.Join(opt.services, ",")
237241
s.cfg.Options.Addr = opt.serverAddr
242+
s.cfg.Options.GRPCAddr = opt.grpcAddr
238243
s.cfg.Options.HTTPRedirectAddr = opt.httpRedirectAddr
239244

240245
ports, err := netutil.AllocatePorts(8)
@@ -293,6 +298,8 @@ func (s *allCmdParam) makeBootstrapConfig(opt allCmdOptions) error {
293298
s.cfg.Options.HTTP3AdvertisePort = null.NewUint32(443, true)
294299
s.cfg.Options.SSHAddr = opt.sshAddr
295300

301+
opt.DataBrokerOptions.apply(&s.cfg)
302+
296303
return nil
297304
}
298305

@@ -306,23 +313,6 @@ func (s *allCmdParam) runConfigControllers(ctx context.Context, cfg *config.Conf
306313
return c.Run(ctx)
307314
}
308315

309-
func getDataBrokerConnection(ctx context.Context, cfg *config.Config) (*grpc.ClientConn, error) {
310-
sharedSecret, err := base64.StdEncoding.DecodeString(cfg.Options.SharedKey)
311-
if err != nil {
312-
return nil, fmt.Errorf("decode shared_secret: %w", err)
313-
}
314-
315-
return grpcutil.NewGRPCClientConn(ctx, &grpcutil.Options{
316-
Address: &url.URL{
317-
Scheme: "http",
318-
Host: net.JoinHostPort("localhost", cfg.GRPCPort),
319-
},
320-
ServiceName: "databroker",
321-
SignedJWTKey: sharedSecret,
322-
RequestTimeout: defaultGRPCTimeout,
323-
})
324-
}
325-
326316
func (s *allCmdParam) buildController(ctx context.Context, cfg *config.Config) (*controllers.Controller, error) {
327317
scheme, err := getScheme()
328318
if err != nil {

‎cmd/databroker_options.go‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package cmd
2+
3+
import (
4+
"context"
5+
"encoding/base64"
6+
"fmt"
7+
"net"
8+
"net/url"
9+
10+
"github.com/spf13/pflag"
11+
"github.com/volatiletech/null/v9"
12+
"google.golang.org/grpc"
13+
"gopkg.in/yaml.v3"
14+
15+
"github.com/pomerium/pomerium/config"
16+
"github.com/pomerium/pomerium/pkg/grpcutil"
17+
)
18+
19+
type dataBrokerOptions struct {
20+
ClusterNodeID string
21+
ClusterNodes dataBrokerClusterNodes
22+
RaftBindAddress string `validate:"omitempty,hostname_port"`
23+
ServiceURLs []string `validate:"dive,url"`
24+
}
25+
26+
// dataBrokerClusterNodes is a custom type to support reading cluster nodes from yaml
27+
type dataBrokerClusterNodes []config.DataBrokerClusterNode
28+
29+
func (n dataBrokerClusterNodes) MarshalText() ([]byte, error) {
30+
return yaml.Marshal([]config.DataBrokerClusterNode(n))
31+
}
32+
33+
func (n *dataBrokerClusterNodes) UnmarshalText(text []byte) error {
34+
return yaml.Unmarshal(text, (*[]config.DataBrokerClusterNode)(n))
35+
}
36+
37+
func (o *dataBrokerOptions) setupFlags(flags *pflag.FlagSet) {
38+
flags.StringVar(&o.ClusterNodeID, "databroker-cluster-node-id", "", "the databroker cluster node id")
39+
flags.TextVar(&o.ClusterNodes, "databroker-cluster-nodes", dataBrokerClusterNodes(nil), "the databroker cluster nodes")
40+
flags.StringVar(&o.RaftBindAddress, "databroker-raft-bind-address", "", "the databroker raft bind address")
41+
flags.StringSliceVar(&o.ServiceURLs, "databroker-service-urls", nil, "the databroker service urls, defaults to localhost")
42+
}
43+
44+
func (o *dataBrokerOptions) apply(dst *config.Config) {
45+
if o.ClusterNodeID != "" {
46+
dst.Options.DataBroker.ClusterNodeID = null.StringFrom(o.ClusterNodeID)
47+
}
48+
dst.Options.DataBroker.ClusterNodes = config.DataBrokerClusterNodes(o.ClusterNodes)
49+
if o.RaftBindAddress != "" {
50+
dst.Options.DataBroker.RaftBindAddress = null.StringFrom(o.RaftBindAddress)
51+
}
52+
dst.Options.DataBroker.ServiceURLs = o.ServiceURLs
53+
}
54+
55+
func getDataBrokerConnection(ctx context.Context, cfg *config.Config) (*grpc.ClientConn, error) {
56+
sharedSecret, err := base64.StdEncoding.DecodeString(cfg.Options.SharedKey)
57+
if err != nil {
58+
return nil, fmt.Errorf("decode shared_secret: %w", err)
59+
}
60+
61+
// use the local gRPC port
62+
dataBrokerURL := &url.URL{Scheme: "http", Host: net.JoinHostPort("127.0.0.1", cfg.GRPCPort)}
63+
// if a databroker service url is set, use the outbound port instead
64+
if len(cfg.Options.DataBroker.ServiceURLs) > 0 || cfg.Options.DataBroker.ServiceURL != "" {
65+
dataBrokerURL = &url.URL{Scheme: "http", Host: net.JoinHostPort("127.0.0.1", cfg.OutboundPort)}
66+
}
67+
68+
return grpcutil.NewGRPCClientConn(ctx, &grpcutil.Options{
69+
Address: dataBrokerURL,
70+
ServiceName: "databroker",
71+
SignedJWTKey: sharedSecret,
72+
RequestTimeout: defaultGRPCTimeout,
73+
})
74+
}

‎cmd/databroker_options_test.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package cmd
2+
3+
import (
4+
"testing"
5+
6+
"github.com/go-playground/validator/v10"
7+
"github.com/spf13/pflag"
8+
"github.com/stretchr/testify/assert"
9+
"github.com/volatiletech/null/v9"
10+
)
11+
12+
func TestDataBrokerOptions(t *testing.T) {
13+
t.Parallel()
14+
15+
parseOptions := func(arguments []string) *dataBrokerOptions {
16+
flags := pflag.NewFlagSet("test", pflag.PanicOnError)
17+
options := &dataBrokerOptions{}
18+
options.setupFlags(flags)
19+
err := flags.Parse(arguments)
20+
assert.NoError(t, err)
21+
return options
22+
}
23+
24+
options := parseOptions([]string{
25+
"--databroker-cluster-node-id", "node-2",
26+
"--databroker-cluster-nodes", `[
27+
{ "id": "node-0", "grpc_address": "127.0.0.1:15000", "raft_address": "127.0.0.1:15100" },
28+
{ "id": "node-1", "grpc_address": "127.0.0.1:15001", "raft_address": "127.0.0.1:15101" },
29+
{ "id": "node-2", "grpc_address": "127.0.0.1:15002", "raft_address": "127.0.0.1:15102" }
30+
]`,
31+
"--databroker-raft-bind-address=:15001",
32+
"--databroker-service-urls=https://databroker-1.example.com",
33+
"--databroker-service-urls=https://databroker-2.example.com",
34+
"--databroker-service-urls=https://databroker-3.example.com",
35+
})
36+
assert.Equal(t, &dataBrokerOptions{
37+
ClusterNodeID: "node-2",
38+
ClusterNodes: dataBrokerClusterNodes{
39+
{ID: "node-0", GRPCAddress: "127.0.0.1:15000", RaftAddress: null.StringFrom("127.0.0.1:15100")},
40+
{ID: "node-1", GRPCAddress: "127.0.0.1:15001", RaftAddress: null.StringFrom("127.0.0.1:15101")},
41+
{ID: "node-2", GRPCAddress: "127.0.0.1:15002", RaftAddress: null.StringFrom("127.0.0.1:15102")},
42+
},
43+
RaftBindAddress: ":15001",
44+
ServiceURLs: []string{
45+
"https://databroker-1.example.com",
46+
"https://databroker-2.example.com",
47+
"https://databroker-3.example.com",
48+
},
49+
}, options)
50+
assert.NoError(t, validator.New().Struct(options))
51+
52+
assert.NoError(t, validator.New().Struct(parseOptions([]string{})))
53+
assert.Error(t, validator.New().Struct(parseOptions([]string{
54+
"--databroker-raft-bind-address", "<INVALID>",
55+
})))
56+
assert.Error(t, validator.New().Struct(parseOptions([]string{
57+
"--databroker-service-urls", "<INVALID>",
58+
})))
59+
}

‎config/crd/bases/ingress.pomerium.io_pomerium.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,14 @@ spec:
189189
- none
190190
type: string
191191
type: object
192+
dataBroker:
193+
description: DataBroker sets the databroker settings.
194+
properties:
195+
clusterLeaderId:
196+
description: ClusterLeaderID defines the cluster leader in a clustered
197+
databroker.
198+
type: string
199+
type: object
192200
dns:
193201
description: DNS sets the dns settings.
194202
properties:

‎deployment.yaml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -331,6 +331,14 @@ spec:
331331
- none
332332
type: string
333333
type: object
334+
dataBroker:
335+
description: DataBroker sets the databroker settings.
336+
properties:
337+
clusterLeaderId:
338+
description: ClusterLeaderID defines the cluster leader in a clustered
339+
databroker.
340+
type: string
341+
type: object
334342
dns:
335343
description: DNS sets the dns settings.
336344
properties:

0 commit comments

Comments
 (0)