-
Notifications
You must be signed in to change notification settings - Fork 1.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* destination: Fix GetProfile endpoint buffering (#11815) In 71635cb and 357a1d3 we updated the endpoint and profile translators to prevent backpressure from stalling the informer tasks. This change updates the endpoint profile translator with the same fix, so that updates are buffered and can detect when when a gRPC stream is stalled. Furthermore, the update method is updated to use a protobuf-aware comparison method instead of using serialization and string comparison. A test is added for the endpoint profile translator, since none existed previously. * stable-2.14.8 This stable release fixes an issue in the control plane where discovery for pod IP addresses could hang indefinitely ([#11815]). --------- Co-authored-by: Oliver Gould <[email protected]>
- Loading branch information
Showing
14 changed files
with
255 additions
and
47 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
127 changes: 127 additions & 0 deletions
127
controller/api/destination/endpoint_profile_translator_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package destination | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| pb "github.com/linkerd/linkerd2-proxy-api/go/destination" | ||
| "github.com/linkerd/linkerd2/controller/api/destination/watcher" | ||
| consts "github.com/linkerd/linkerd2/pkg/k8s" | ||
| logging "github.com/sirupsen/logrus" | ||
| corev1 "k8s.io/api/core/v1" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| ) | ||
|
|
||
| func TestEndpointProfileTranslator(t *testing.T) { | ||
| // logging.SetLevel(logging.TraceLevel) | ||
| // defer logging.SetLevel(logging.PanicLevel) | ||
|
|
||
| addr := &watcher.Address{ | ||
| IP: "10.10.11.11", | ||
| Port: 8080, | ||
| } | ||
| podAddr := &watcher.Address{ | ||
| IP: "10.10.11.11", | ||
| Port: 8080, | ||
| Pod: &corev1.Pod{ | ||
| ObjectMeta: metav1.ObjectMeta{ | ||
| Annotations: map[string]string{ | ||
| consts.ProxyOpaquePortsAnnotation: "8080", | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| t.Run("Sends update", func(t *testing.T) { | ||
| mockGetProfileServer := &mockDestinationGetProfileServer{ | ||
| profilesReceived: make(chan *pb.DestinationProfile, 1), | ||
| } | ||
| log := logging.WithField("test", t.Name()) | ||
| translator := newEndpointProfileTranslator( | ||
| true, "cluster", "identity", make(map[uint32]struct{}), | ||
| nil, nil, | ||
| mockGetProfileServer, | ||
| nil, | ||
| log, | ||
| ) | ||
| translator.Start() | ||
| defer translator.Stop() | ||
|
|
||
| if err := translator.Update(addr); err != nil { | ||
| t.Fatal("Expected update") | ||
| } | ||
| select { | ||
| case p := <-mockGetProfileServer.profilesReceived: | ||
| log.Debugf("Received update: %v", p) | ||
| case <-time.After(1 * time.Second): | ||
| t.Fatal("No update received") | ||
| } | ||
|
|
||
| if err := translator.Update(addr); err != nil { | ||
| t.Fatal("Unexpected update") | ||
| } | ||
| select { | ||
| case p := <-mockGetProfileServer.profilesReceived: | ||
| t.Fatalf("Duplicate update sent: %v", p) | ||
| case <-time.After(1 * time.Second): | ||
| } | ||
|
|
||
| if err := translator.Update(podAddr); err != nil { | ||
| t.Fatal("Expected update") | ||
| } | ||
| select { | ||
| case p := <-mockGetProfileServer.profilesReceived: | ||
| log.Debugf("Received update: %v", p) | ||
| case <-time.After(1 * time.Second): | ||
| } | ||
| }) | ||
|
|
||
| t.Run("Handles overflow", func(t *testing.T) { | ||
| mockGetProfileServer := &mockDestinationGetProfileServer{ | ||
| profilesReceived: make(chan *pb.DestinationProfile, 1), | ||
| } | ||
| log := logging.WithField("test", t.Name()) | ||
| endStream := make(chan struct{}) | ||
| translator := newEndpointProfileTranslator( | ||
| true, "cluster", "identity", make(map[uint32]struct{}), | ||
| nil, nil, | ||
| mockGetProfileServer, | ||
| endStream, | ||
| log, | ||
| ) | ||
| translator.Start() | ||
| defer translator.Stop() | ||
|
|
||
| for i := 0; i < updateQueueCapacity/2; i++ { | ||
| if err := translator.Update(podAddr); err != nil { | ||
| t.Fatal("Expected update") | ||
| } | ||
| select { | ||
| case <-endStream: | ||
| t.Fatal("Stream ended prematurely") | ||
| default: | ||
| } | ||
|
|
||
| if err := translator.Update(addr); err != nil { | ||
| t.Fatal("Expected update") | ||
| } | ||
| select { | ||
| case <-endStream: | ||
| t.Fatal("Stream ended prematurely") | ||
| default: | ||
| } | ||
| } | ||
|
|
||
| if err := translator.Update(podAddr); err == nil { | ||
| t.Fatal("Expected update to fail") | ||
| } | ||
| select { | ||
| case <-endStream: | ||
| default: | ||
| t.Fatal("Stream should have ended") | ||
| } | ||
|
|
||
| // XXX We should assert that endpointProfileUpdatesQueueOverflowCounter | ||
| // == 1 but we can't read counter values. | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.