-
Notifications
You must be signed in to change notification settings - Fork 2.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[extension/k8s leader elector] Extension leader election implementati…
…on (#38015) <!--Ex. Fixing a bug - Describe the bug and how this fixes the issue. Ex. Adding a feature - Explain what this achieves.--> #### Description Added the actual implementation for leader election Initial structure was added here: [basic structure](#37266) <!-- Issue number (e.g. #1234) or full URL to issue, if applicable. --> #### Link to tracking issue Fixes #34460 <!--Describe what testing was performed and which tests were added.--> #### Testing unit tests added <!--Describe the documentation added.--> #### Documentation documentation is provided <!--Please delete paragraphs that you did not use before submitting.-->
- Loading branch information
1 parent
20a49b7
commit f036dd9
Showing
7 changed files
with
228 additions
and
9 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package k8sleaderelector | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| "go.opentelemetry.io/collector/component/componenttest" | ||
| "go.uber.org/zap" | ||
| "go.uber.org/zap/zaptest/observer" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/kubernetes/fake" | ||
| "k8s.io/utils/ptr" | ||
|
|
||
| "github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig" | ||
| ) | ||
|
|
||
| func TestExtension(t *testing.T) { | ||
| config := &Config{ | ||
| LeaseName: "foo", | ||
| LeaseNamespace: "default", | ||
| LeaseDuration: 15 * time.Second, | ||
| RenewDuration: 10 * time.Second, | ||
| RetryPeriod: 2 * time.Second, | ||
| } | ||
|
|
||
| iamInvokedOnLeading := false | ||
|
|
||
| ctx := context.TODO() | ||
| fakeClient := fake.NewClientset() | ||
| config.makeClient = func(_ k8sconfig.APIConfig) (kubernetes.Interface, error) { | ||
| return fakeClient, nil | ||
| } | ||
|
|
||
| observedZapCore, _ := observer.New(zap.WarnLevel) | ||
|
|
||
| leaderElection := leaderElectionExtension{ | ||
| config: config, | ||
| client: fakeClient, | ||
| logger: zap.New(observedZapCore), | ||
| leaseHolderID: "foo", | ||
| } | ||
|
|
||
| leaderElection.SetCallBackFuncs( | ||
| func(_ context.Context) { | ||
| iamInvokedOnLeading = true | ||
| fmt.Printf("LeaderElection started leading") | ||
| }, | ||
| func() { | ||
| fmt.Printf("LeaderElection stopped leading") | ||
| }, | ||
| ) | ||
|
|
||
| require.NoError(t, leaderElection.Start(ctx, componenttest.NewNopHost())) | ||
|
|
||
| expectedLeaseDurationSeconds := ptr.To(int32(15)) | ||
|
|
||
| require.Eventually(t, func() bool { | ||
| lease, err := fakeClient.CoordinationV1().Leases("default").Get(ctx, "foo", metav1.GetOptions{}) | ||
| require.NoError(t, err) | ||
| require.NotNil(t, lease) | ||
| require.Equal(t, expectedLeaseDurationSeconds, lease.Spec.LeaseDurationSeconds) | ||
| return true | ||
| }, 10*time.Second, 100*time.Millisecond) | ||
|
|
||
| require.True(t, iamInvokedOnLeading) | ||
| require.NoError(t, leaderElection.Shutdown(ctx)) | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package k8sleaderelector // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/k8sleaderelector" | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/tools/leaderelection" | ||
| "k8s.io/client-go/tools/leaderelection/resourcelock" | ||
| ) | ||
|
|
||
| func newK8sLeaderElector( | ||
| cfg *Config, | ||
| client kubernetes.Interface, | ||
| onStartedLeading func(context.Context), | ||
| onStoppedLeading func(), | ||
| identity string, | ||
| ) (*leaderelection.LeaderElector, error) { | ||
| resourceLock, err := resourcelock.New( | ||
| resourcelock.LeasesResourceLock, | ||
| cfg.LeaseNamespace, | ||
| cfg.LeaseName, | ||
| client.CoreV1(), | ||
| client.CoordinationV1(), | ||
| resourcelock.ResourceLockConfig{ | ||
| Identity: identity, | ||
| }) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| leConfig := leaderelection.LeaderElectionConfig{ | ||
| Lock: resourceLock, | ||
| LeaseDuration: cfg.LeaseDuration, | ||
| RenewDeadline: cfg.RenewDuration, | ||
| RetryPeriod: cfg.RetryPeriod, | ||
| Callbacks: leaderelection.LeaderCallbacks{ | ||
| OnStartedLeading: onStartedLeading, | ||
| OnStoppedLeading: onStoppedLeading, | ||
| }, | ||
| } | ||
|
|
||
| return leaderelection.NewLeaderElector(leConfig) | ||
| } |
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,30 @@ | ||
| package k8sleaderelector | ||
|
|
||
| // Copyright The OpenTelemetry Authors | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| "k8s.io/client-go/kubernetes/fake" | ||
| ) | ||
|
|
||
| func TestLeaderElector(t *testing.T) { | ||
| fakeClient := fake.NewClientset() | ||
| onStartedLeading := func(_ context.Context) {} | ||
| onStoppedLeading := func() {} | ||
| leConfig := Config{ | ||
| LeaseName: "foo", | ||
| LeaseNamespace: "bar", | ||
| LeaseDuration: 20 * time.Second, | ||
| RenewDuration: 10 * time.Second, | ||
| RetryPeriod: 2 * time.Second, | ||
| } | ||
|
|
||
| leaderElector, err := newK8sLeaderElector(&leConfig, fakeClient, onStartedLeading, onStoppedLeading, "host1") | ||
| require.NoError(t, err) | ||
| require.NotNil(t, leaderElector) | ||
| } |