8913011676

Committed 01 May 2024 06:01PM UTC coverage: 57.367% (-0.01%) from 57.377%

Build # 8913011676

Build Type

push

github

Committed by

web-flow

Commit Message

chore(deps): bump the github-actions group with 5 updates (#5094)

Bumps the github-actions group with 5 updates:

| Package | From | To |
| --- | --- | --- |
| [actions/checkout](https://github.com/actions/checkout) | `4.1.2` | `4.1.4` |
| [docker/setup-buildx-action](https://github.com/docker/setup-buildx-action) | `3.2.0` | `3.3.0` |
| [stefanzweifel/git-auto-commit-action](https://github.com/stefanzweifel/git-auto-commit-action) | `5.0.0` | `5.0.1` |
| [golangci/golangci-lint-action](https://github.com/golangci/golangci-lint-action) | `4.0.0` | `5.1.0` |
| [actions/upload-artifact](https://github.com/actions/upload-artifact) | `4.3.1` | `4.3.3` |


Updates `actions/checkout` from 4.1.2 to 4.1.4
- [Release notes](https://github.com/actions/checkout/releases)
- [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md)
- [Commits](https://github.com/actions/checkout/compare/9bb56186c...0ad4b8fad)

Updates `docker/setup-buildx-action` from 3.2.0 to 3.3.0
- [Release notes](https://github.com/docker/setup-buildx-action/releases)
- [Commits](https://github.com/docker/setup-buildx-action/compare/2b5128504...d70bba72b)

Updates `stefanzweifel/git-auto-commit-action` from 5.0.0 to 5.0.1
- [Release notes](https://github.com/stefanzweifel/git-auto-commit-action/releases)
- [Changelog](https://github.com/stefanzweifel/git-auto-commit-action/blob/master/CHANGELOG.md)
- [Commits](https://github.com/stefanzweifel/git-auto-commit-action/compare/8756aa072...8621497c8)

Updates `golangci/golangci-lint-action` from 4.0.0 to 5.1.0
- [Release notes](https://github.com/golangci/golangci-lint-action/releases)
- [Commits](https://github.com/golangci/golangci-lint-action/compare/3cfe3a4ab...9d1e0624a798bb64f6c3cea93db47... (continued)

Run Details

16770 of 29233 relevant lines covered (57.37%)

103.82 hits per line

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

75.46

/internal/controlplane/xdsmgr/xdsmgr.go

// Package xdsmgr implements a resource discovery manager for envoy.
package xdsmgr

import (
        "context"
        "sync"

        envoy_service_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
        "github.com/google/uuid"
        "golang.org/x/sync/errgroup"
        "google.golang.org/grpc/codes"
        "google.golang.org/grpc/status"

        "github.com/pomerium/pomerium/internal/log"
        "github.com/pomerium/pomerium/internal/signal"
)

type streamState struct {
        typeURL                string
        clientResourceVersions map[string]string
        unsubscribedResources  map[string]struct{}
}

var onHandleDeltaRequest = func(state *streamState) {}

// A Manager manages xDS resources.
type Manager struct {
        signal *signal.Signal

        mu        sync.Mutex
        nonce     string
        resources map[string][]*envoy_service_discovery_v3.Resource
}

// NewManager creates a new Manager.
func NewManager(resources map[string][]*envoy_service_discovery_v3.Resource) *Manager {
        return &Manager{
                signal: signal.New(),

                nonce:     uuid.New().String(),
                resources: resources,
        }
}

// DeltaAggregatedResources implements the increment xDS server.
func (mgr *Manager) DeltaAggregatedResources(
        stream envoy_service_discovery_v3.AggregatedDiscoveryService_DeltaAggregatedResourcesServer,
) error {
        ch := mgr.signal.Bind()
        defer mgr.signal.Unbind(ch)

        stateByTypeURL := map[string]*streamState{}

        getDeltaResponse := func(ctx context.Context, typeURL string) *envoy_service_discovery_v3.DeltaDiscoveryResponse {
                mgr.mu.Lock()
                defer mgr.mu.Unlock()

                state, ok := stateByTypeURL[typeURL]
                if !ok {
                        return nil
                }

                res := &envoy_service_discovery_v3.DeltaDiscoveryResponse{
                        TypeUrl: typeURL,
                        Nonce:   mgr.nonce,
                }
                seen := map[string]struct{}{}
                for _, resource := range mgr.resources[typeURL] {
                        seen[resource.Name] = struct{}{}
                        if resource.Version != state.clientResourceVersions[resource.Name] {
                                res.Resources = append(res.Resources, resource)
                        }
                }
                for name := range state.clientResourceVersions {
                        _, ok := seen[name]
                        if !ok {
                                res.RemovedResources = append(res.RemovedResources, name)
                        }
                }

                if len(res.Resources) == 0 && len(res.RemovedResources) == 0 {
                        return nil
                }

                return res
        }

        handleDeltaRequest := func(ctx context.Context, req *envoy_service_discovery_v3.DeltaDiscoveryRequest) {
                mgr.mu.Lock()
                defer mgr.mu.Unlock()

                state, ok := stateByTypeURL[req.GetTypeUrl()]
                if !ok {
                        // first time we've seen a message for this type URL.
                        state = &streamState{
                                typeURL:                req.GetTypeUrl(),
                                clientResourceVersions: req.GetInitialResourceVersions(),
                                unsubscribedResources:  make(map[string]struct{}),
                        }
                        if state.clientResourceVersions == nil {
                                state.clientResourceVersions = make(map[string]string)
                        }
                        stateByTypeURL[req.GetTypeUrl()] = state
                }

                switch {
                case req.GetResponseNonce() == "":
                        // neither an ACK or a NACK
                case req.GetErrorDetail() != nil:
                        // a NACK
                        // - set the client resource versions to the current resource versions
                        state.clientResourceVersions = make(map[string]string)
                        for _, resource := range mgr.resources[req.GetTypeUrl()] {
                                state.clientResourceVersions[resource.Name] = resource.Version
                        }
                        logNACK(req)
                case req.GetResponseNonce() == mgr.nonce:
                        // an ACK for the last response
                        // - set the client resource versions to the current resource versions
                        state.clientResourceVersions = make(map[string]string)
                        for _, resource := range mgr.resources[req.GetTypeUrl()] {
                                state.clientResourceVersions[resource.Name] = resource.Version
                        }
                        logACK(req)
                default:
                        // an ACK for a response that's not the last response
                        log.Debug(ctx).
                                Str("type-url", req.GetTypeUrl()).
                                Msg("xdsmgr: ack")
                }

                // update subscriptions
                for _, name := range req.GetResourceNamesSubscribe() {
                        delete(state.unsubscribedResources, name)
                }
                for _, name := range req.GetResourceNamesUnsubscribe() {
                        state.unsubscribedResources[name] = struct{}{}
                        // from the docs:
                        //   NOTE: the server must respond with all resources listed in
                        //   resource_names_subscribe, even if it believes the client has
                        //   the most recent version of them. The reason: the client may
                        //   have dropped them, but then regained interest before it had
                        //   a chance to send the unsubscribe message.
                        // so we reset the version to treat it like a new version
                        delete(state.clientResourceVersions, name)
                }

                onHandleDeltaRequest(state)
        }

        incoming := make(chan *envoy_service_discovery_v3.DeltaDiscoveryRequest)
        outgoing := make(chan *envoy_service_discovery_v3.DeltaDiscoveryResponse)
        eg, ctx := errgroup.WithContext(stream.Context())
        // 1. receive all incoming messages
        eg.Go(func() error {
                for {
                        req, err := stream.Recv()
                        if err != nil {
                                return err
                        }

                        select {
                        case <-ctx.Done():
                                return ctx.Err()
                        case incoming <- req:
                        }
                }
        })
        // 2. handle incoming requests or resource changes
        eg.Go(func() error {
                changeCtx := ctx
                for {
                        var typeURLs []string
                        select {
                        case <-ctx.Done():
                                return ctx.Err()
                        case req := <-incoming:
                                handleDeltaRequest(changeCtx, req)
                                typeURLs = []string{req.GetTypeUrl()}
                        case changeCtx = <-ch:
                                mgr.mu.Lock()
                                for typeURL := range mgr.resources {
                                        typeURLs = append(typeURLs, typeURL)
                                }
                                mgr.mu.Unlock()
                        }

                        for _, typeURL := range typeURLs {
                                res := getDeltaResponse(changeCtx, typeURL)
                                if res == nil {
                                        continue
                                }

                                select {
                                case <-ctx.Done():
                                        return ctx.Err()
                                case outgoing <- res:
                                }
                        }
                }
        })
        // 3. send all outgoing messages
        eg.Go(func() error {
                for {
                        select {
                        case <-ctx.Done():
                                return ctx.Err()
                        case res := <-outgoing:
                                log.Debug(ctx).
                                        Str("type-url", res.GetTypeUrl()).
                                        Int("resource-count", len(res.GetResources())).
                                        Int("removed-resource-count", len(res.GetRemovedResources())).
                                        Msg("xdsmgr: sending resources")
                                err := stream.Send(res)
                                if err != nil {
                                        return err
                                }
                        }
                }
        })
        return eg.Wait()
}

// StreamAggregatedResources is not implemented.
func (mgr *Manager) StreamAggregatedResources(
        _ envoy_service_discovery_v3.AggregatedDiscoveryService_StreamAggregatedResourcesServer,
) error {
        return status.Errorf(codes.Unimplemented, "method StreamAggregatedResources not implemented")
}

// Update updates the state of resources. If any changes are made they will be pushed to any listening
// streams. For each TypeURL the list of resources should be the complete list of resources.
func (mgr *Manager) Update(ctx context.Context, resources map[string][]*envoy_service_discovery_v3.Resource) {
        nonce := uuid.New().String()

        mgr.mu.Lock()
        mgr.nonce = nonce
        mgr.resources = resources
        mgr.mu.Unlock()

        mgr.signal.Broadcast(ctx)
}

1	// Package xdsmgr implements a resource discovery manager for envoy.
2	package xdsmgr
3
4	import (
5	"context"
6	"sync"
7
8	envoy_service_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
9	"github.com/google/uuid"
10	"golang.org/x/sync/errgroup"
11	"google.golang.org/grpc/codes"
12	"google.golang.org/grpc/status"
13
14	"github.com/pomerium/pomerium/internal/log"
15	"github.com/pomerium/pomerium/internal/signal"
16	)
17
18	type streamState struct {
19	typeURL string
20	clientResourceVersions map[string]string
21	unsubscribedResources map[string]struct{}
22	}
23
24	var onHandleDeltaRequest = func(state *streamState) {}	×
25
26	// A Manager manages xDS resources.
27	type Manager struct {
28	signal *signal.Signal
29
30	mu sync.Mutex
31	nonce string
32	resources map[string][]*envoy_service_discovery_v3.Resource
33	}
34
35	// NewManager creates a new Manager.
36	func NewManager(resources map[string][]envoy_service_discovery_v3.Resource) Manager {	1✔
37	return &Manager{	1✔
38	signal: signal.New(),	1✔
39		1✔
40	nonce: uuid.New().String(),	1✔
41	resources: resources,	1✔
42	}	1✔
43	}	1✔
44
45	// DeltaAggregatedResources implements the increment xDS server.
46	func (mgr *Manager) DeltaAggregatedResources(
47	stream envoy_service_discovery_v3.AggregatedDiscoveryService_DeltaAggregatedResourcesServer,
48	) error {	1✔
49	ch := mgr.signal.Bind()	1✔
50	defer mgr.signal.Unbind(ch)	1✔
51		1✔
52	stateByTypeURL := map[string]*streamState{}	1✔
53		1✔
54	getDeltaResponse := func(ctx context.Context, typeURL string) *envoy_service_discovery_v3.DeltaDiscoveryResponse {	7✔
55	mgr.mu.Lock()	6✔
56	defer mgr.mu.Unlock()	6✔
57		6✔
58	state, ok := stateByTypeURL[typeURL]	6✔
59	if !ok {	6✔
60	return nil	×
61	}	×
62
63	res := &envoy_service_discovery_v3.DeltaDiscoveryResponse{	6✔
64	TypeUrl: typeURL,	6✔
65	Nonce: mgr.nonce,	6✔
66	}	6✔
67	seen := map[string]struct{}{}	6✔
68	for _, resource := range mgr.resources[typeURL] {	10✔
69	seen[resource.Name] = struct{}{}	4✔
70	if resource.Version != state.clientResourceVersions[resource.Name] {	6✔
71	res.Resources = append(res.Resources, resource)	2✔
72	}	2✔
73	}
74	for name := range state.clientResourceVersions {	10✔
75	_, ok := seen[name]	4✔
76	if !ok {	5✔
77	res.RemovedResources = append(res.RemovedResources, name)	1✔
78	}	1✔
79	}
80
81	if len(res.Resources) == 0 && len(res.RemovedResources) == 0 {	9✔
82	return nil	3✔
83	}	3✔
84
85	return res	3✔
86	}
87
88	handleDeltaRequest := func(ctx context.Context, req *envoy_service_discovery_v3.DeltaDiscoveryRequest) {	5✔
89	mgr.mu.Lock()	4✔
90	defer mgr.mu.Unlock()	4✔
91		4✔
92	state, ok := stateByTypeURL[req.GetTypeUrl()]	4✔
93	if !ok {	5✔
94	// first time we've seen a message for this type URL.	1✔
95	state = &streamState{	1✔
96	typeURL: req.GetTypeUrl(),	1✔
97	clientResourceVersions: req.GetInitialResourceVersions(),	1✔
98	unsubscribedResources: make(map[string]struct{}),	1✔
99	}	1✔
100	if state.clientResourceVersions == nil {	2✔
101	state.clientResourceVersions = make(map[string]string)	1✔
102	}	1✔
103	stateByTypeURL[req.GetTypeUrl()] = state	1✔
104	}
105
106	switch {	4✔
107	case req.GetResponseNonce() == "":	1✔
108	// neither an ACK or a NACK
109	case req.GetErrorDetail() != nil:	×
110	// a NACK	×
111	// - set the client resource versions to the current resource versions	×
112	state.clientResourceVersions = make(map[string]string)	×
113	for _, resource := range mgr.resources[req.GetTypeUrl()] {	×
114	state.clientResourceVersions[resource.Name] = resource.Version	×
115	}	×
116	logNACK(req)	×
117	case req.GetResponseNonce() == mgr.nonce:	3✔
118	// an ACK for the last response	3✔
119	// - set the client resource versions to the current resource versions	3✔
120	state.clientResourceVersions = make(map[string]string)	3✔
121	for _, resource := range mgr.resources[req.GetTypeUrl()] {	5✔
122	state.clientResourceVersions[resource.Name] = resource.Version	2✔
123	}	2✔
124	logACK(req)	3✔
125	default:	×
126	// an ACK for a response that's not the last response	×
127	log.Debug(ctx).	×
128	Str("type-url", req.GetTypeUrl()).	×
129	Msg("xdsmgr: ack")	×
130	}
131
132	// update subscriptions
133	for _, name := range req.GetResourceNamesSubscribe() {	4✔
134	delete(state.unsubscribedResources, name)	×
135	}	×
136	for _, name := range req.GetResourceNamesUnsubscribe() {	4✔
137	state.unsubscribedResources[name] = struct{}{}	×
138	// from the docs:	×
139	// NOTE: the server must respond with all resources listed in	×
140	// resource_names_subscribe, even if it believes the client has	×
141	// the most recent version of them. The reason: the client may	×
142	// have dropped them, but then regained interest before it had	×
143	// a chance to send the unsubscribe message.	×
144	// so we reset the version to treat it like a new version	×
145	delete(state.clientResourceVersions, name)	×
146	}	×
147
148	onHandleDeltaRequest(state)	4✔
149	}
150
151	incoming := make(chan *envoy_service_discovery_v3.DeltaDiscoveryRequest)	1✔
152	outgoing := make(chan *envoy_service_discovery_v3.DeltaDiscoveryResponse)	1✔
153	eg, ctx := errgroup.WithContext(stream.Context())	1✔
154	// 1. receive all incoming messages	1✔
155	eg.Go(func() error {	2✔
156	for {	6✔
157	req, err := stream.Recv()	5✔
158	if err != nil {	5✔
159	return err	×
160	}	×
161
162	select {	4✔
163	case <-ctx.Done():	×
164	return ctx.Err()	×
165	case incoming <- req:	4✔
166	}
167	}
168	})
169	// 2. handle incoming requests or resource changes
170	eg.Go(func() error {	2✔
171	changeCtx := ctx	1✔
172	for {	8✔
173	var typeURLs []string	7✔
174	select {	7✔
175	case <-ctx.Done():	×
176	return ctx.Err()	×
177	case req := <-incoming:	4✔
178	handleDeltaRequest(changeCtx, req)	4✔
179	typeURLs = []string{req.GetTypeUrl()}	4✔
180	case changeCtx = <-ch:	2✔
181	mgr.mu.Lock()	2✔
182	for typeURL := range mgr.resources {	4✔
183	typeURLs = append(typeURLs, typeURL)	2✔
184	}	2✔
185	mgr.mu.Unlock()	2✔
186	}
187
188	for _, typeURL := range typeURLs {	12✔
189	res := getDeltaResponse(changeCtx, typeURL)	6✔
190	if res == nil {	9✔
191	continue	3✔
192	}
193
194	select {	3✔
195	case <-ctx.Done():	×
196	return ctx.Err()	×
197	case outgoing <- res:	3✔
198	}
199	}
200	}
201	})
202	// 3. send all outgoing messages
203	eg.Go(func() error {	2✔
204	for {	5✔
205	select {	4✔
206	case <-ctx.Done():	×
207	return ctx.Err()	×
208	case res := <-outgoing:	3✔
209	log.Debug(ctx).	3✔
210	Str("type-url", res.GetTypeUrl()).	3✔
211	Int("resource-count", len(res.GetResources())).	3✔
212	Int("removed-resource-count", len(res.GetRemovedResources())).	3✔
213	Msg("xdsmgr: sending resources")	3✔
214	err := stream.Send(res)	3✔
215	if err != nil {	3✔
216	return err	×
217	}	×
218	}
219	}
220	})
221	return eg.Wait()	1✔
222	}
223
224	// StreamAggregatedResources is not implemented.
225	func (mgr *Manager) StreamAggregatedResources(
226	_ envoy_service_discovery_v3.AggregatedDiscoveryService_StreamAggregatedResourcesServer,
227	) error {	1✔
228	return status.Errorf(codes.Unimplemented, "method StreamAggregatedResources not implemented")	1✔
229	}	1✔
230
231	// Update updates the state of resources. If any changes are made they will be pushed to any listening
232	// streams. For each TypeURL the list of resources should be the complete list of resources.
233	func (mgr Manager) Update(ctx context.Context, resources map[string][]envoy_service_discovery_v3.Resource) {	2✔
234	nonce := uuid.New().String()	2✔
235		2✔
236	mgr.mu.Lock()	2✔
237	mgr.nonce = nonce	2✔
238	mgr.resources = resources	2✔
239	mgr.mu.Unlock()	2✔
240		2✔
241	mgr.signal.Broadcast(ctx)	2✔
242	}	2✔

pomerium / pomerium / 8913011676

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