Skip to content

Commit

Permalink
Don't request ES to clear routing allocation exclude at every reconci…
Browse files Browse the repository at this point in the history 
…liation (#2610)

* Don't clear shard allocation excludes in every reconciliation

This commits stores the last updated value of cluster routing allocation
exclude in an annotation of the Elasticsearch resource.
Before doing any call to the ES API for that setting, we check the
existing annotation. If its value is the same as the one we're about to
set, skip the ES API call.
If not, do the call then store the setting value in the annotation.

The annotation can be removed at any time: it will just trigger a call
to the Elasticsearch API with the right expected value, and the
annotation will be set again.

Users can manipulate this setting behind our back, in which case the
operator will only react based on the annotation value. We consider it's
the user responsibility to not mess cluster settings up at this point.

* Fix comment/function name mismatch
  • Loading branch information
@sebgl
sebgl authored Mar 2, 2020
1 parent 4ee5ae5 commit d010f2c
Show file tree
Hide file tree
Showing 3 changed files with 156 additions and 11 deletions.
5 changes: 1 addition & 4 deletions pkg/controller/elasticsearch/driver/downscale.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,7 @@ func HandleDownscale(
// migrate data away from nodes that should be removed
// if leavingNodes is empty, it clears any existing settings
leavingNodes := leavingNodeNames(downscales)
if len(leavingNodes) != 0 {
log.V(1).Info("Migrating data away from nodes", "nodes", leavingNodes)
}
if err := migration.MigrateData(downscaleCtx.parentCtx, downscaleCtx.esClient, leavingNodes); err != nil {
if err := migration.MigrateData(downscaleCtx.parentCtx, downscaleCtx.k8sClient, downscaleCtx.es, downscaleCtx.esClient, leavingNodes); err != nil {
return results.WithError(err)
}

Expand Down
51 changes: 48 additions & 3 deletions pkg/controller/elasticsearch/migration/migrate_data.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,20 @@ import (
"context"
"strings"

logf "sigs.k8s.io/controller-runtime/pkg/log"

esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
esclient "github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
)

var log = logf.Log.WithName("migrate-data")

const (
// AllocationExcludeAnnotationName is the name of the annotation that stores the last
// cluster.routing.allocation._name setting applied to the Elasticsearch cluster.
AllocationExcludeAnnotationName = "elasticsearch.k8s.elastic.co/allocation-exclude"
)

func shardIsMigrating(toMigrate client.Shard, others []client.Shard) bool {
Expand Down Expand Up @@ -76,12 +88,45 @@ func IsMigratingData(ctx context.Context, shardLister esclient.ShardLister, podN
return nodeIsMigratingData(podName, shards, excludedNodes), nil
}

// allocationExcludeFromAnnotation returns the allocation exclude value stored in an annotation.
// May be empty if not set.
func allocationExcludeFromAnnotation(es esv1.Elasticsearch) string {
return es.Annotations[AllocationExcludeAnnotationName]
}

// updateAllocationExcludeAnnotation sets an annotation in ES with the given cluster routing allocation exclude value.
// This is to avoid making the same ES API call over and over again.
func updateAllocationExcludeAnnotation(c k8s.Client, es esv1.Elasticsearch, value string) error {
if es.Annotations == nil {
es.Annotations = map[string]string{}
}
es.Annotations[AllocationExcludeAnnotationName] = value
return c.Update(&es)
}

// MigrateData sets allocation filters for the given nodes.
func MigrateData(ctx context.Context, allocationSetter esclient.AllocationSetter, leavingNodes []string) error {
func MigrateData(
ctx context.Context,
c k8s.Client,
es esv1.Elasticsearch,
allocationSetter esclient.AllocationSetter,
leavingNodes []string,
) error {
// compute the expected exclusion value
exclusions := "none_excluded"
if len(leavingNodes) > 0 {
exclusions = strings.Join(leavingNodes, ",")
}
// update allocation exclusions
return allocationSetter.ExcludeFromShardAllocation(ctx, exclusions)
// compare with what was set previously
// Note the user may have changed it behind our back through the ES API. It is considered their responsibility.
// Manually removing the annotation to force a refresh of the allocations exclude setting is a valid use case.
if exclusions == allocationExcludeFromAnnotation(es) {
return nil
}
log.Info("Setting routing allocation excludes", "namespace", es.Namespace, "es_name", es.Name, "value", exclusions)
if err := allocationSetter.ExcludeFromShardAllocation(ctx, exclusions); err != nil {
return err
}
// store updated value in an annotation so we don't make the same call over and over again
return updateAllocationExcludeAnnotation(c, es, exclusions)
}
111 changes: 107 additions & 4 deletions pkg/controller/elasticsearch/migration/migrate_data_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,12 @@ import (
"fmt"
"testing"

metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

esv1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
"github.com/elastic/cloud-on-k8s/pkg/controller/elasticsearch/client"
"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
Expand Down Expand Up @@ -138,32 +143,130 @@ func TestIsMigratingData(t *testing.T) {
func TestMigrateData(t *testing.T) {
tests := []struct {
name string
es esv1.Elasticsearch
leavingNodes []string
want string
wantEs esv1.Elasticsearch
}{
{
name: "no nodes to migrate",
name: "no nodes to migrate, no annotation on ES",
es: esv1.Elasticsearch{},
leavingNodes: []string{},
want: "none_excluded",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
},
{
name: "no nodes to migrate, annotation already set on ES",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
leavingNodes: []string{},
want: "",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
},
{
name: "no nodes to migrate, annotation set with some exclusions on ES",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node2"},
}},
leavingNodes: []string{},
want: "none_excluded",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
},
{
name: "one node to migrate, no annotation set on ES",
es: esv1.Elasticsearch{},
leavingNodes: []string{"test-node"},
want: "test-node",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node"},
}},
},
{
name: "one node to migrate, no exclusions in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
leavingNodes: []string{"test-node"},
want: "test-node",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node"},
}},
},
{
name: "one node to migrate",
name: "one node to migrate, different exclusions in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node2"},
}},
leavingNodes: []string{"test-node"},
want: "test-node",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node"},
}},
},
{
name: "one node to migrate, already present in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node"},
}},
leavingNodes: []string{"test-node"},
want: "",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node"},
}},
},
{
name: "multiple node to migrate, no exclusions in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "none_excluded"},
}},
leavingNodes: []string{"test-node1", "test-node2"},
want: "test-node1,test-node2",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node2"},
}},
},
{
name: "multiple node to migrate",
name: "multiple node to migrate, different exclusions in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node3"},
}},
leavingNodes: []string{"test-node1", "test-node2"},
want: "test-node1,test-node2",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node2"},
}},
},
{
name: "multiple node to migrate, already present in ES annotation",
es: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node2"},
}},
leavingNodes: []string{"test-node1", "test-node2"},
want: "",
wantEs: esv1.Elasticsearch{ObjectMeta: metav1.ObjectMeta{
Annotations: map[string]string{AllocationExcludeAnnotationName: "test-node1,test-node2"},
}},
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
allocationSetter := fakeAllocationSetter{}
err := MigrateData(context.Background(), &allocationSetter, tt.leavingNodes)
c := k8s.WrappedFakeClient(&tt.es)
err := MigrateData(context.Background(), c, tt.es, &allocationSetter, tt.leavingNodes)
require.NoError(t, err)
assert.Contains(t, allocationSetter.value, tt.want)
var retrievedES esv1.Elasticsearch
err = c.Get(k8s.ExtractNamespacedName(&tt.es), &retrievedES)
require.NoError(t, err)
require.Equal(t, tt.wantEs.Annotations, retrievedES.Annotations)
})
}
}

0 comments on commit d010f2c

Please sign in to comment.