5284233483

Committed 15 Jun 2023 10:34PM UTC coverage: 63.521% (+0.05%) from 63.469%

Build # 5284233483

Build Type

push

github

Committed by

web-flow

Commit Message

chore(deps): bump golangci/golangci-lint-action from 3.5.0 to 3.6.0 (#4238)

Bumps [golangci/golangci-lint-action](https://github.com/golangci/golangci-lint-action) from 3.5.0 to 3.6.0.
- [Release notes](https://github.com/golangci/golangci-lint-action/releases)
- [Commits](https://github.com/golangci/golangci-lint-action/compare/5f1fec701...639cd343e)

---
updated-dependencies:
- dependency-name: golangci/golangci-lint-action
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

13535 of 21308 relevant lines covered (63.52%)

123.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.46

/internal/databroker/config_source.go

package databroker

import (
        "context"
        "sort"
        "sync"
        "time"

        "golang.org/x/exp/maps"

        "github.com/pomerium/pomerium/config"
        "github.com/pomerium/pomerium/internal/hashutil"
        "github.com/pomerium/pomerium/internal/log"
        "github.com/pomerium/pomerium/internal/telemetry/metrics"
        "github.com/pomerium/pomerium/internal/telemetry/trace"
        "github.com/pomerium/pomerium/pkg/cryptutil"
        "github.com/pomerium/pomerium/pkg/grpc"
        configpb "github.com/pomerium/pomerium/pkg/grpc/config"
        "github.com/pomerium/pomerium/pkg/grpc/databroker"
        "github.com/pomerium/pomerium/pkg/grpcutil"
)

// ConfigSource provides a new Config source that decorates an underlying config with
// configuration derived from the data broker.
type ConfigSource struct {
        mu                     sync.RWMutex
        outboundGRPCConnection *grpc.CachedOutboundGRPClientConn
        computedConfig         *config.Config
        underlyingConfig       *config.Config
        dbConfigs              map[string]dbConfig
        updaterHash            uint64
        cancel                 func()

        config.ChangeDispatcher
}

type dbConfig struct {
        *configpb.Config
        version uint64
}

// NewConfigSource creates a new ConfigSource.
func NewConfigSource(ctx context.Context, underlying config.Source, listeners ...config.ChangeListener) *ConfigSource {
        src := &ConfigSource{
                dbConfigs:              map[string]dbConfig{},
                outboundGRPCConnection: new(grpc.CachedOutboundGRPClientConn),
        }
        for _, li := range listeners {
                src.OnConfigChange(ctx, li)
        }
        underlying.OnConfigChange(ctx, func(ctx context.Context, cfg *config.Config) {
                src.mu.Lock()
                src.underlyingConfig = cfg.Clone()
                src.mu.Unlock()

                src.rebuild(ctx, firstTime(false))
        })
        src.underlyingConfig = underlying.GetConfig()
        src.rebuild(ctx, firstTime(true))
        return src
}

// GetConfig gets the current config.
func (src *ConfigSource) GetConfig() *config.Config {
        src.mu.RLock()
        defer src.mu.RUnlock()

        return src.computedConfig
}

type firstTime bool

func (src *ConfigSource) rebuild(ctx context.Context, firstTime firstTime) {
        _, span := trace.StartSpan(ctx, "databroker.config_source.rebuild")
        defer span.End()

        src.mu.Lock()
        defer src.mu.Unlock()

        cfg := src.underlyingConfig.Clone()

        // start the updater
        src.runUpdater(cfg)

        seen := map[uint64]string{}
        for _, policy := range cfg.Options.GetAllPolicies() {
                id, err := policy.RouteID()
                if err != nil {
                        log.Warn(ctx).Err(err).
                                Str("policy", policy.String()).
                                Msg("databroker: invalid policy config, ignoring")
                        return
                }
                seen[id] = ""
        }

        var additionalPolicies []config.Policy

        ids := maps.Keys(src.dbConfigs)
        sort.Strings(ids)

        certsIndex := cryptutil.NewCertificatesIndex()
        for _, cert := range cfg.Options.GetX509Certificates() {
                certsIndex.Add(cert)
        }

        // add all the config policies to the list
        for _, id := range ids {
                cfgpb := src.dbConfigs[id]

                cfg.Options.ApplySettings(ctx, certsIndex, cfgpb.Settings)
                var errCount uint64

                err := cfg.Options.Validate()
                if err != nil {
                        metrics.SetDBConfigRejected(ctx, cfg.Options.Services, id, cfgpb.version, err)
                        return
                }

                for _, routepb := range cfgpb.GetRoutes() {
                        policy, err := config.NewPolicyFromProto(routepb)
                        if err != nil {
                                errCount++
                                log.Warn(ctx).Err(err).
                                        Str("db_config_id", id).
                                        Msg("databroker: error converting protobuf into policy")
                                continue
                        }

                        err = policy.Validate()
                        if err != nil {
                                errCount++
                                log.Warn(ctx).Err(err).
                                        Str("db_config_id", id).
                                        Str("policy", policy.String()).
                                        Msg("databroker: invalid policy, ignoring")
                                continue
                        }

                        routeID, err := policy.RouteID()
                        if err != nil {
                                errCount++
                                log.Warn(ctx).Err(err).
                                        Str("db_config_id", id).
                                        Str("policy", policy.String()).
                                        Msg("databroker: cannot establish policy route ID, ignoring")
                                continue
                        }

                        if _, ok := seen[routeID]; ok {
                                errCount++
                                log.Warn(ctx).Err(err).
                                        Str("db_config_id", id).
                                        Str("seen-in", seen[routeID]).
                                        Str("policy", policy.String()).
                                        Msg("databroker: duplicate policy detected, ignoring")
                                continue
                        }
                        seen[routeID] = id

                        additionalPolicies = append(additionalPolicies, *policy)
                }
                metrics.SetDBConfigInfo(ctx, cfg.Options.Services, id, cfgpb.version, int64(errCount))
        }

        // add the additional policies here since calling `Validate` will reset them.
        cfg.Options.AdditionalPolicies = append(cfg.Options.AdditionalPolicies, additionalPolicies...)

        src.computedConfig = cfg
        if !firstTime {
                src.Trigger(ctx, cfg)
        }

        metrics.SetConfigInfo(ctx, cfg.Options.Services, "databroker", cfg.Checksum(), true)
}

func (src *ConfigSource) runUpdater(cfg *config.Config) {
        sharedKey, _ := cfg.Options.GetSharedKey()
        connectionOptions := &grpc.OutboundOptions{
                OutboundPort:   cfg.OutboundPort,
                InstallationID: cfg.Options.InstallationID,
                ServiceName:    cfg.Options.Services,
                SignedJWTKey:   sharedKey,
        }
        h, err := hashutil.Hash(connectionOptions)
        if err != nil {
                log.Fatal().Err(err).Send()
        }
        // nothing changed, so don't restart the updater
        if src.updaterHash == h {
                return
        }
        src.updaterHash = h

        if src.cancel != nil {
                src.cancel()
                src.cancel = nil
        }

        ctx := context.Background()
        ctx, src.cancel = context.WithCancel(ctx)

        cc, err := src.outboundGRPCConnection.Get(ctx, connectionOptions)
        if err != nil {
                log.Error(ctx).Err(err).Msg("databroker: failed to create gRPC connection to data broker")
                return
        }

        client := databroker.NewDataBrokerServiceClient(cc)

        syncer := databroker.NewSyncer("databroker", &syncerHandler{
                client: client,
                src:    src,
        }, databroker.WithTypeURL(grpcutil.GetTypeURL(new(configpb.Config))),
                databroker.WithFastForward())
        go func() {
                var databrokerURLs []string
                urls, _ := cfg.Options.GetDataBrokerURLs()
                for _, url := range urls {
                        databrokerURLs = append(databrokerURLs, url.String())
                }

                log.Info(ctx).
                        Str("outbound_port", cfg.OutboundPort).
                        Strs("databroker_urls", databrokerURLs).
                        Msg("config: starting databroker config source syncer")
                _ = grpc.WaitForReady(ctx, cc, time.Second*10)
                _ = syncer.Run(ctx)
        }()
}

type syncerHandler struct {
        src    *ConfigSource
        client databroker.DataBrokerServiceClient
}

func (s *syncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
        return s.client
}

func (s *syncerHandler) ClearRecords(_ context.Context) {
        s.src.mu.Lock()
        s.src.dbConfigs = map[string]dbConfig{}
        s.src.mu.Unlock()
}

func (s *syncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) {
        if len(records) == 0 {
                return
        }

        s.src.mu.Lock()
        for _, record := range records {
                if record.GetDeletedAt() != nil {
                        delete(s.src.dbConfigs, record.GetId())
                        continue
                }

                var cfgpb configpb.Config
                err := record.GetData().UnmarshalTo(&cfgpb)
                if err != nil {
                        log.Warn(ctx).Err(err).Msg("databroker: error decoding config")
                        delete(s.src.dbConfigs, record.GetId())
                        continue
                }

                s.src.dbConfigs[record.GetId()] = dbConfig{&cfgpb, record.Version}
        }
        s.src.mu.Unlock()

        s.src.rebuild(ctx, firstTime(false))
}

1	package databroker
2
3	import (
4	"context"
5	"sort"
6	"sync"
7	"time"
8
9	"golang.org/x/exp/maps"
10
11	"github.com/pomerium/pomerium/config"
12	"github.com/pomerium/pomerium/internal/hashutil"
13	"github.com/pomerium/pomerium/internal/log"
14	"github.com/pomerium/pomerium/internal/telemetry/metrics"
15	"github.com/pomerium/pomerium/internal/telemetry/trace"
16	"github.com/pomerium/pomerium/pkg/cryptutil"
17	"github.com/pomerium/pomerium/pkg/grpc"
18	configpb "github.com/pomerium/pomerium/pkg/grpc/config"
19	"github.com/pomerium/pomerium/pkg/grpc/databroker"
20	"github.com/pomerium/pomerium/pkg/grpcutil"
21	)
22
23	// ConfigSource provides a new Config source that decorates an underlying config with
24	// configuration derived from the data broker.
25	type ConfigSource struct {
26	mu sync.RWMutex
27	outboundGRPCConnection *grpc.CachedOutboundGRPClientConn
28	computedConfig *config.Config
29	underlyingConfig *config.Config
30	dbConfigs map[string]dbConfig
31	updaterHash uint64
32	cancel func()
33
34	config.ChangeDispatcher
35	}
36
37	type dbConfig struct {
38	*configpb.Config
39	version uint64
40	}
41
42	// NewConfigSource creates a new ConfigSource.
43	func NewConfigSource(ctx context.Context, underlying config.Source, listeners ...config.ChangeListener) *ConfigSource {	1✔
44	src := &ConfigSource{	1✔
45	dbConfigs: map[string]dbConfig{},	1✔
46	outboundGRPCConnection: new(grpc.CachedOutboundGRPClientConn),	1✔
47	}	1✔
48	for _, li := range listeners {	2✔
49	src.OnConfigChange(ctx, li)	1✔
50	}	1✔
51	underlying.OnConfigChange(ctx, func(ctx context.Context, cfg *config.Config) {	2✔
52	src.mu.Lock()	1✔
53	src.underlyingConfig = cfg.Clone()	1✔
54	src.mu.Unlock()	1✔
55		1✔
56	src.rebuild(ctx, firstTime(false))	1✔
57	})	1✔
58	src.underlyingConfig = underlying.GetConfig()	1✔
59	src.rebuild(ctx, firstTime(true))	1✔
60	return src	1✔
61	}
62
63	// GetConfig gets the current config.
64	func (src ConfigSource) GetConfig() config.Config {	1✔
65	src.mu.RLock()	1✔
66	defer src.mu.RUnlock()	1✔
67		1✔
68	return src.computedConfig	1✔
69	}	1✔
70
71	type firstTime bool
72
73	func (src *ConfigSource) rebuild(ctx context.Context, firstTime firstTime) {	3✔
74	_, span := trace.StartSpan(ctx, "databroker.config_source.rebuild")	3✔
75	defer span.End()	3✔
76		3✔
77	src.mu.Lock()	3✔
78	defer src.mu.Unlock()	3✔
79		3✔
80	cfg := src.underlyingConfig.Clone()	3✔
81		3✔
82	// start the updater	3✔
83	src.runUpdater(cfg)	3✔
84		3✔
85	seen := map[uint64]string{}	3✔
86	for _, policy := range cfg.Options.GetAllPolicies() {	6✔
87	id, err := policy.RouteID()	3✔
88	if err != nil {	3✔
89	log.Warn(ctx).Err(err).	×
90	Str("policy", policy.String()).	×
91	Msg("databroker: invalid policy config, ignoring")	×
92	return	×
93	}	×
94	seen[id] = ""	3✔
95	}
96
97	var additionalPolicies []config.Policy	3✔
98		3✔
99	ids := maps.Keys(src.dbConfigs)	3✔
100	sort.Strings(ids)	3✔
101		3✔
102	certsIndex := cryptutil.NewCertificatesIndex()	3✔
103	for _, cert := range cfg.Options.GetX509Certificates() {	6✔
104	certsIndex.Add(cert)	3✔
105	}	3✔
106
107	// add all the config policies to the list
108	for _, id := range ids {	5✔
109	cfgpb := src.dbConfigs[id]	2✔
110		2✔
111	cfg.Options.ApplySettings(ctx, certsIndex, cfgpb.Settings)	2✔
112	var errCount uint64	2✔
113		2✔
114	err := cfg.Options.Validate()	2✔
115	if err != nil {	2✔
116	metrics.SetDBConfigRejected(ctx, cfg.Options.Services, id, cfgpb.version, err)	×
117	return	×
118	}	×
119
120	for _, routepb := range cfgpb.GetRoutes() {	4✔
121	policy, err := config.NewPolicyFromProto(routepb)	2✔
122	if err != nil {	2✔
123	errCount++	×
124	log.Warn(ctx).Err(err).	×
125	Str("db_config_id", id).	×
126	Msg("databroker: error converting protobuf into policy")	×
127	continue	×
128	}
129
130	err = policy.Validate()	2✔
131	if err != nil {	2✔
132	errCount++	×
133	log.Warn(ctx).Err(err).	×
134	Str("db_config_id", id).	×
135	Str("policy", policy.String()).	×
136	Msg("databroker: invalid policy, ignoring")	×
137	continue	×
138	}
139
140	routeID, err := policy.RouteID()	2✔
141	if err != nil {	2✔
142	errCount++	×
143	log.Warn(ctx).Err(err).	×
144	Str("db_config_id", id).	×
145	Str("policy", policy.String()).	×
146	Msg("databroker: cannot establish policy route ID, ignoring")	×
147	continue	×
148	}
149
150	if _, ok := seen[routeID]; ok {	2✔
151	errCount++	×
152	log.Warn(ctx).Err(err).	×
153	Str("db_config_id", id).	×
154	Str("seen-in", seen[routeID]).	×
155	Str("policy", policy.String()).	×
156	Msg("databroker: duplicate policy detected, ignoring")	×
157	continue	×
158	}
159	seen[routeID] = id	2✔
160		2✔
161	additionalPolicies = append(additionalPolicies, *policy)	2✔
162	}
163	metrics.SetDBConfigInfo(ctx, cfg.Options.Services, id, cfgpb.version, int64(errCount))	2✔
164	}
165
166	// add the additional policies here since calling `Validate` will reset them.
167	cfg.Options.AdditionalPolicies = append(cfg.Options.AdditionalPolicies, additionalPolicies...)	3✔
168		3✔
169	src.computedConfig = cfg	3✔
170	if !firstTime {	5✔
171	src.Trigger(ctx, cfg)	2✔
172	}	2✔
173
174	metrics.SetConfigInfo(ctx, cfg.Options.Services, "databroker", cfg.Checksum(), true)	3✔
175	}
176
177	func (src ConfigSource) runUpdater(cfg config.Config) {	3✔
178	sharedKey, _ := cfg.Options.GetSharedKey()	3✔
179	connectionOptions := &grpc.OutboundOptions{	3✔
180	OutboundPort: cfg.OutboundPort,	3✔
181	InstallationID: cfg.Options.InstallationID,	3✔
182	ServiceName: cfg.Options.Services,	3✔
183	SignedJWTKey: sharedKey,	3✔
184	}	3✔
185	h, err := hashutil.Hash(connectionOptions)	3✔
186	if err != nil {	3✔
187	log.Fatal().Err(err).Send()	×
188	}	×
189	// nothing changed, so don't restart the updater
190	if src.updaterHash == h {	5✔
191	return	2✔
192	}	2✔
193	src.updaterHash = h	1✔
194		1✔
195	if src.cancel != nil {	1✔
196	src.cancel()	×
197	src.cancel = nil	×
198	}	×
199
200	ctx := context.Background()	1✔
201	ctx, src.cancel = context.WithCancel(ctx)	1✔
202		1✔
203	cc, err := src.outboundGRPCConnection.Get(ctx, connectionOptions)	1✔
204	if err != nil {	1✔
205	log.Error(ctx).Err(err).Msg("databroker: failed to create gRPC connection to data broker")	×
206	return	×
207	}	×
208
209	client := databroker.NewDataBrokerServiceClient(cc)	1✔
210		1✔
211	syncer := databroker.NewSyncer("databroker", &syncerHandler{	1✔
212	client: client,	1✔
213	src: src,	1✔
214	}, databroker.WithTypeURL(grpcutil.GetTypeURL(new(configpb.Config))),	1✔
215	databroker.WithFastForward())	1✔
216	go func() {	2✔
217	var databrokerURLs []string	1✔
218	urls, _ := cfg.Options.GetDataBrokerURLs()	1✔
219	for _, url := range urls {	2✔
220	databrokerURLs = append(databrokerURLs, url.String())	1✔
221	}	1✔
222
223	log.Info(ctx).	1✔
224	Str("outbound_port", cfg.OutboundPort).	1✔
225	Strs("databroker_urls", databrokerURLs).	1✔
226	Msg("config: starting databroker config source syncer")	1✔
227	_ = grpc.WaitForReady(ctx, cc, time.Second*10)	1✔
228	_ = syncer.Run(ctx)	1✔
229	}()
230	}
231
232	type syncerHandler struct {
233	src *ConfigSource
234	client databroker.DataBrokerServiceClient
235	}
236
237	func (s *syncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {	2✔
238	return s.client	2✔
239	}	2✔
240
241	func (s *syncerHandler) ClearRecords(_ context.Context) {	1✔
242	s.src.mu.Lock()	1✔
243	s.src.dbConfigs = map[string]dbConfig{}	1✔
244	s.src.mu.Unlock()	1✔
245	}	1✔
246
247	func (s syncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []databroker.Record) {	1✔
248	if len(records) == 0 {	1✔
249	return	×
250	}	×
251
252	s.src.mu.Lock()	1✔
253	for _, record := range records {	2✔
254	if record.GetDeletedAt() != nil {	1✔
255	delete(s.src.dbConfigs, record.GetId())	×
256	continue	×
257	}
258
259	var cfgpb configpb.Config	1✔
260	err := record.GetData().UnmarshalTo(&cfgpb)	1✔
261	if err != nil {	1✔
262	log.Warn(ctx).Err(err).Msg("databroker: error decoding config")	×
263	delete(s.src.dbConfigs, record.GetId())	×
264	continue	×
265	}
266
267	s.src.dbConfigs[record.GetId()] = dbConfig{&cfgpb, record.Version}	1✔
268	}
269	s.src.mu.Unlock()	1✔
270		1✔
271	s.src.rebuild(ctx, firstTime(false))	1✔
272	}

pomerium / pomerium / 5284233483

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous