cassandra.go

Documentation: k8s.io/kubernetes/test/e2e/upgrades/apps

     1  /*
     2  Copyright 2017 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package apps
    18  
    19  import (
    20  	"context"
    21  	"encoding/json"
    22  	"fmt"
    23  	"io"
    24  	"net"
    25  	"net/http"
    26  	"path/filepath"
    27  	"sync"
    28  	"time"
    29  
    30  	"github.com/onsi/ginkgo/v2"
    31  	"github.com/onsi/gomega"
    32  
    33  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    34  	"k8s.io/apimachinery/pkg/util/version"
    35  	"k8s.io/apimachinery/pkg/util/wait"
    36  
    37  	"k8s.io/kubernetes/test/e2e/framework"
    38  	e2ekubectl "k8s.io/kubernetes/test/e2e/framework/kubectl"
    39  	e2estatefulset "k8s.io/kubernetes/test/e2e/framework/statefulset"
    40  	e2etestfiles "k8s.io/kubernetes/test/e2e/framework/testfiles"
    41  	"k8s.io/kubernetes/test/e2e/upgrades"
    42  )
    43  
    44  const cassandraManifestPath = "test/e2e/testing-manifests/statefulset/cassandra"
    45  
    46  // CassandraUpgradeTest ups and verifies that a Cassandra StatefulSet behaves
    47  // well across upgrades.
    48  type CassandraUpgradeTest struct {
    49  	ip               string
    50  	successfulWrites int
    51  }
    52  
    53  // Name returns the tracking name of the test.
    54  func (CassandraUpgradeTest) Name() string { return "cassandra-upgrade" }
    55  
    56  // Skip returns true when this test can be skipped.
    57  func (CassandraUpgradeTest) Skip(upgCtx upgrades.UpgradeContext) bool {
    58  	minVersion := version.MustParseSemantic("1.6.0")
    59  	for _, vCtx := range upgCtx.Versions {
    60  		if vCtx.Version.LessThan(minVersion) {
    61  			return true
    62  		}
    63  	}
    64  	return false
    65  }
    66  
    67  func cassandraKubectlCreate(ns, file string) {
    68  	data, err := e2etestfiles.Read(filepath.Join(cassandraManifestPath, file))
    69  	if err != nil {
    70  		framework.Fail(err.Error())
    71  	}
    72  	input := string(data)
    73  	e2ekubectl.RunKubectlOrDieInput(ns, input, "create", "-f", "-")
    74  }
    75  
    76  // Setup creates a Cassandra StatefulSet and a PDB. It also brings up a tester
    77  // ReplicaSet and associated service and PDB to guarantee availability during
    78  // the upgrade.
    79  // It waits for the system to stabilize before adding two users to verify
    80  // connectivity.
    81  func (t *CassandraUpgradeTest) Setup(ctx context.Context, f *framework.Framework) {
    82  	ns := f.Namespace.Name
    83  	statefulsetPoll := 30 * time.Second
    84  	statefulsetTimeout := 10 * time.Minute
    85  
    86  	ginkgo.By("Creating a PDB")
    87  	cassandraKubectlCreate(ns, "pdb.yaml")
    88  
    89  	ginkgo.By("Creating a Cassandra StatefulSet")
    90  	e2estatefulset.CreateStatefulSet(ctx, f.ClientSet, cassandraManifestPath, ns)
    91  
    92  	ginkgo.By("Creating a cassandra-test-server deployment")
    93  	cassandraKubectlCreate(ns, "tester.yaml")
    94  
    95  	ginkgo.By("Getting the ingress IPs from the services")
    96  	err := wait.PollUntilContextTimeout(ctx, statefulsetPoll, statefulsetTimeout, true, func(ctx context.Context) (bool, error) {
    97  		if t.ip = t.getServiceIP(ctx, f, ns, "test-server"); t.ip == "" {
    98  			return false, nil
    99  		}
   100  		if _, err := t.listUsers(); err != nil {
   101  			framework.Logf("Service endpoint is up but isn't responding")
   102  			return false, nil
   103  		}
   104  		return true, nil
   105  	})
   106  	framework.ExpectNoError(err)
   107  	framework.Logf("Service endpoint is up")
   108  
   109  	ginkgo.By("Adding 2 dummy users")
   110  	err = t.addUser("Alice")
   111  	framework.ExpectNoError(err)
   112  	err = t.addUser("Bob")
   113  	framework.ExpectNoError(err)
   114  	t.successfulWrites = 2
   115  
   116  	ginkgo.By("Verifying that the users exist")
   117  	users, err := t.listUsers()
   118  	framework.ExpectNoError(err)
   119  	gomega.Expect(users).To(gomega.HaveLen(2))
   120  }
   121  
   122  // listUsers gets a list of users from the db via the tester service.
   123  func (t *CassandraUpgradeTest) listUsers() ([]string, error) {
   124  	r, err := http.Get(fmt.Sprintf("http://%s/list", net.JoinHostPort(t.ip, "8080")))
   125  	if err != nil {
   126  		return nil, err
   127  	}
   128  	defer r.Body.Close()
   129  	if r.StatusCode != http.StatusOK {
   130  		b, err := io.ReadAll(r.Body)
   131  		if err != nil {
   132  			return nil, err
   133  		}
   134  		return nil, fmt.Errorf(string(b))
   135  	}
   136  	var names []string
   137  	if err := json.NewDecoder(r.Body).Decode(&names); err != nil {
   138  		return nil, err
   139  	}
   140  	return names, nil
   141  }
   142  
   143  // addUser adds a user to the db via the tester services.
   144  func (t *CassandraUpgradeTest) addUser(name string) error {
   145  	val := map[string][]string{"name": {name}}
   146  	r, err := http.PostForm(fmt.Sprintf("http://%s/add", net.JoinHostPort(t.ip, "8080")), val)
   147  	if err != nil {
   148  		return err
   149  	}
   150  	defer r.Body.Close()
   151  	if r.StatusCode != http.StatusOK {
   152  		b, err := io.ReadAll(r.Body)
   153  		if err != nil {
   154  			return err
   155  		}
   156  		return fmt.Errorf(string(b))
   157  	}
   158  	return nil
   159  }
   160  
   161  // getServiceIP is a helper method to extract the Ingress IP from the service.
   162  func (t *CassandraUpgradeTest) getServiceIP(ctx context.Context, f *framework.Framework, ns, svcName string) string {
   163  	svc, err := f.ClientSet.CoreV1().Services(ns).Get(ctx, svcName, metav1.GetOptions{})
   164  	framework.ExpectNoError(err)
   165  	ingress := svc.Status.LoadBalancer.Ingress
   166  	if len(ingress) == 0 {
   167  		return ""
   168  	}
   169  	return ingress[0].IP
   170  }
   171  
   172  // Test is called during the upgrade.
   173  // It launches two goroutines, one continuously writes to the db and one reads
   174  // from the db. Each attempt is tallied and at the end we verify if the success
   175  // ratio is over a certain threshold (0.75). We also verify that we get
   176  // at least the same number of rows back as we successfully wrote.
   177  func (t *CassandraUpgradeTest) Test(ctx context.Context, f *framework.Framework, done <-chan struct{}, upgrade upgrades.UpgradeType) {
   178  	ginkgo.By("Continuously polling the database during upgrade.")
   179  	var (
   180  		success, failures, writeAttempts, lastUserCount int
   181  		mu                                              sync.Mutex
   182  		errors                                          = map[string]int{}
   183  	)
   184  	// Write loop.
   185  	go wait.Until(func() {
   186  		writeAttempts++
   187  		if err := t.addUser(fmt.Sprintf("user-%d", writeAttempts)); err != nil {
   188  			framework.Logf("Unable to add user: %v", err)
   189  			mu.Lock()
   190  			errors[err.Error()]++
   191  			mu.Unlock()
   192  			return
   193  		}
   194  		t.successfulWrites++
   195  	}, 10*time.Millisecond, done)
   196  	// Read loop.
   197  	wait.Until(func() {
   198  		users, err := t.listUsers()
   199  		if err != nil {
   200  			framework.Logf("Could not retrieve users: %v", err)
   201  			failures++
   202  			mu.Lock()
   203  			errors[err.Error()]++
   204  			mu.Unlock()
   205  			return
   206  		}
   207  		success++
   208  		lastUserCount = len(users)
   209  	}, 10*time.Millisecond, done)
   210  	framework.Logf("got %d users; want >=%d", lastUserCount, t.successfulWrites)
   211  	gomega.Expect(lastUserCount).To(gomega.BeNumerically(">=", t.successfulWrites), "lastUserCount is too small")
   212  	ratio := float64(success) / float64(success+failures)
   213  	framework.Logf("Successful gets %d/%d=%v", success, success+failures, ratio)
   214  	ratio = float64(t.successfulWrites) / float64(writeAttempts)
   215  	framework.Logf("Successful writes %d/%d=%v", t.successfulWrites, writeAttempts, ratio)
   216  	framework.Logf("Errors: %v", errors)
   217  	// TODO(maisem): tweak this value once we have a few test runs.
   218  	gomega.Expect(ratio).To(gomega.BeNumerically(">", 0.75), "ratio too small")
   219  }
   220  
   221  // Teardown does one final check of the data's availability.
   222  func (t *CassandraUpgradeTest) Teardown(ctx context.Context, f *framework.Framework) {
   223  	users, err := t.listUsers()
   224  	framework.ExpectNoError(err)
   225  	gomega.Expect(len(users)).To(gomega.BeNumerically(">=", t.successfulWrites), "len(users) is too small")
   226  }
   227
View as plain text