Skip to content

Commit

Permalink
Backport of remove all k8s services from consul into release/1.5.x (#…
Browse files Browse the repository at this point in the history 
…4298)

* backport of commit 0c5150d

* backport of commit e3b74dd

* backport of commit 0a9a7cb

* backport of commit a211357

* backport of commit 45381f9

* backport of commit e5061e4

* backport of commit 27b4eb9

* backport of commit 82ff2a9

---------

Co-authored-by: Xinyi Wang <[email protected]>
  • Loading branch information
@hc-github-team-consul-core @xwa153
hc-github-team-consul-core and xwa153 authored Sep 3, 2024
1 parent 5caba50 commit 82ed2e7
Show file tree
Hide file tree
Showing 3 changed files with 284 additions and 21 deletions.
3 changes: 3 additions & 0 deletions .changelog/4255.txt
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
```release-note:bug
sync-catalog: Enable the user to purge the registered services by passing parent node and necessary filters.
```
143 changes: 122 additions & 21 deletions control-plane/subcommand/sync-catalog/command.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,19 +4,22 @@
package synccatalog

import (
"bufio"
"context"
"flag"
"fmt"
"net/http"
"os"
"os/signal"
"regexp"
"strings"
"sync"
"syscall"
"time"

mapset "github.com/deckarep/golang-set"
"github.com/hashicorp/consul-server-connection-manager/discovery"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/go-hclog"
"github.com/mitchellh/cli"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand All @@ -37,27 +40,29 @@ import (
type Command struct {
UI cli.Ui

flags *flag.FlagSet
consul *flags.ConsulFlags
k8s *flags.K8SFlags
flagListen string
flagToConsul bool
flagToK8S bool
flagConsulDomain string
flagConsulK8STag string
flagConsulNodeName string
flagK8SDefault bool
flagK8SServicePrefix string
flagConsulServicePrefix string
flagK8SSourceNamespace string
flagK8SWriteNamespace string
flagConsulWritePeriod time.Duration
flagSyncClusterIPServices bool
flagSyncLBEndpoints bool
flagNodePortSyncType string
flagAddK8SNamespaceSuffix bool
flagLogLevel string
flagLogJSON bool
flags *flag.FlagSet
consul *flags.ConsulFlags
k8s *flags.K8SFlags
flagListen string
flagToConsul bool
flagToK8S bool
flagConsulDomain string
flagConsulK8STag string
flagConsulNodeName string
flagK8SDefault bool
flagK8SServicePrefix string
flagConsulServicePrefix string
flagK8SSourceNamespace string
flagK8SWriteNamespace string
flagConsulWritePeriod time.Duration
flagSyncClusterIPServices bool
flagSyncLBEndpoints bool
flagNodePortSyncType string
flagAddK8SNamespaceSuffix bool
flagLogLevel string
flagLogJSON bool
flagPurgeK8SServicesFromNode string
flagFilter string

// Flags to support namespaces
flagEnableNamespaces bool // Use namespacing on all components
Expand Down Expand Up @@ -138,6 +143,10 @@ func (c *Command) init() {
"\"debug\", \"info\", \"warn\", and \"error\".")
c.flags.BoolVar(&c.flagLogJSON, "log-json", false,
"Enable or disable JSON output format for logging.")
c.flags.StringVar(&c.flagPurgeK8SServicesFromNode, "purge-k8s-services-from-node", "",
"Purge all K8S services registered in Consul under the node name.")
c.flags.StringVar(&c.flagFilter, "filter", "",
"Specifies the expression used to filter the queries results for the node.")

c.flags.Var((*flags.AppendSliceValue)(&c.flagAllowK8sNamespacesList), "allow-k8s-namespace",
"K8s namespaces to explicitly allow. May be specified multiple times.")
Expand Down Expand Up @@ -251,6 +260,19 @@ func (c *Command) Run(args []string) int {
}
c.ready = true

if c.flagPurgeK8SServicesFromNode != "" {
consulClient, err := consul.NewClientFromConnMgr(consulConfig, c.connMgr)
if err != nil {
c.UI.Error(fmt.Sprintf("unable to instantiate consul client: %s", err))
return 1
}
if err := c.removeAllK8SServicesFromConsulNode(consulClient, c.flagPurgeK8SServicesFromNode); err != nil {
c.UI.Error(fmt.Sprintf("unable to remove all K8S services: %s", err))
return 1
}
return 0
}

// Convert allow/deny lists to sets
allowSet := flags.ToSet(c.flagAllowK8sNamespacesList)
denySet := flags.ToSet(c.flagDenyK8sNamespacesList)
Expand Down Expand Up @@ -393,6 +415,85 @@ func (c *Command) Run(args []string) int {
}
}

// remove all k8s services from Consul.
func (c *Command) removeAllK8SServicesFromConsulNode(consulClient *api.Client, nodeName string) error {
node, _, err := consulClient.Catalog().NodeServiceList(nodeName, &api.QueryOptions{Filter: c.flagFilter})
if err != nil {
return err
}

var wg sync.WaitGroup
services := node.Services
errChan := make(chan error, 1)
batchSize := 300
maxRetries := 2
retryDelay := 200 * time.Millisecond

// Ask for user confirmation
reader := bufio.NewReader(os.Stdin)
for {
c.UI.Info(fmt.Sprintf("Are you sure you want to delete %v K8S services from %v? (y/n): ", len(services), nodeName))
input, _ := reader.ReadString('\n')
input = strings.TrimSpace(input)
if input == "y" || input == "Y" {
break
} else if input == "n" || input == "N" {
return nil
} else {
c.UI.Info("Invalid input. Please enter 'y' or 'n'.")
}
}

for i := 0; i < len(services); i += batchSize {
end := i + batchSize
if end > len(services) {
end = len(services)
}

wg.Add(1)
go func(batch []*api.AgentService) {
defer wg.Done()

for _, service := range batch {
err := retryOps(func() error {
_, err := consulClient.Catalog().Deregister(&api.CatalogDeregistration{
Node: nodeName,
ServiceID: service.ID,
}, nil)
return err
}, maxRetries, retryDelay, c.logger)
if err != nil {
if len(errChan) == 0 {
errChan <- err
}
}
}
c.UI.Info(fmt.Sprintf("Processed %v K8S services from %v", len(batch), nodeName))
}(services[i:end])
wg.Wait()
}

close(errChan)
if err = <-errChan; err != nil {
return err
}
c.UI.Info("All K8S services were deregistered from Consul")
return nil
}

func retryOps(operation func() error, maxRetries int, retryDelay time.Duration, logger hclog.Logger) error {
var err error
for i := 0; i < maxRetries; i++ {
err = operation()
if err == nil {
return nil
}
logger.Warn("Operation failed: %v. Retrying in %v millisecond...", err, retryDelay)
time.Sleep(retryDelay)
}
return err
}

func (c *Command) handleReady(rw http.ResponseWriter, _ *http.Request) {
if !c.ready {
c.UI.Error("[GET /health/ready] sync catalog controller is not yet ready")
Expand Down
159 changes: 159 additions & 0 deletions control-plane/subcommand/sync-catalog/command_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"testing"
"time"

"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/sdk/testutil/retry"
"github.com/hashicorp/go-hclog"
"github.com/mitchellh/cli"
Expand Down Expand Up @@ -571,6 +572,164 @@ func TestRun_ToConsulChangingFlags(t *testing.T) {
}
}

// Test services could be de-registered from Consul.
func TestRemoveAllK8SServicesFromConsul(t *testing.T) {
t.Parallel()

k8s, testClient := completeSetup(t)
consulClient := testClient.APIClient

// Create a mock reader to simulate user input
input := "y\n"
reader, writer, err := os.Pipe()
require.NoError(t, err)
oldStdin := os.Stdin
os.Stdin = reader
defer func() { os.Stdin = oldStdin }()

// Write the simulated user input to the mock reader
go func() {
defer writer.Close()
_, err := writer.WriteString(input)
require.NoError(t, err)
}()

// Run the command.
ui := cli.NewMockUi()
cmd := Command{
UI: ui,
clientset: k8s,
logger: hclog.New(&hclog.LoggerOptions{
Name: t.Name(),
Level: hclog.Debug,
}),
flagAllowK8sNamespacesList: []string{"*"},
connMgr: testClient.Watcher,
}

// create two services in k8s
_, err = k8s.CoreV1().Services("bar").Create(context.Background(), lbService("foo", "1.1.1.1"), metav1.CreateOptions{})
require.NoError(t, err)

_, err = k8s.CoreV1().Services("baz").Create(context.Background(), lbService("foo", "2.2.2.2"), metav1.CreateOptions{})
require.NoError(t, err)

longRunningChan := runCommandAsynchronously(&cmd, []string{
"-addresses", "127.0.0.1",
"-http-port", strconv.Itoa(testClient.Cfg.HTTPPort),
"-consul-write-interval", "100ms",
"-add-k8s-namespace-suffix",
})
defer stopCommand(t, &cmd, longRunningChan)

// check that the name of the service is namespaced
retry.Run(t, func(r *retry.R) {
svc, _, err := consulClient.Catalog().Service("foo-bar", "k8s", nil)
require.NoError(r, err)
require.Len(r, svc, 1)
require.Equal(r, "1.1.1.1", svc[0].ServiceAddress)
svc, _, err = consulClient.Catalog().Service("foo-baz", "k8s", nil)
require.NoError(r, err)
require.Len(r, svc, 1)
require.Equal(r, "2.2.2.2", svc[0].ServiceAddress)
})

exitChan := runCommandAsynchronously(&cmd, []string{
"-addresses", "127.0.0.1",
"-http-port", strconv.Itoa(testClient.Cfg.HTTPPort),
"-purge-k8s-services-from-node=k8s-sync",
})
stopCommand(t, &cmd, exitChan)

retry.Run(t, func(r *retry.R) {
serviceList, _, err := consulClient.Catalog().NodeServiceList("k8s-sync", &api.QueryOptions{AllowStale: false})
require.NoError(r, err)
require.Len(r, serviceList.Services, 0)
})
}

// Test services could be de-registered from Consul with filter.
func TestRemoveAllK8SServicesFromConsulWithFilter(t *testing.T) {
t.Parallel()

k8s, testClient := completeSetup(t)
consulClient := testClient.APIClient

// Create a mock reader to simulate user input
input := "y\n"
reader, writer, err := os.Pipe()
require.NoError(t, err)
oldStdin := os.Stdin
os.Stdin = reader
defer func() { os.Stdin = oldStdin }()

// Write the simulated user input to the mock reader
go func() {
defer writer.Close()
_, err := writer.WriteString(input)
require.NoError(t, err)
}()

// Run the command.
ui := cli.NewMockUi()
cmd := Command{
UI: ui,
clientset: k8s,
logger: hclog.New(&hclog.LoggerOptions{
Name: t.Name(),
Level: hclog.Debug,
}),
flagAllowK8sNamespacesList: []string{"*"},
connMgr: testClient.Watcher,
}

// create two services in k8s
_, err = k8s.CoreV1().Services("bar").Create(context.Background(), lbService("foo", "1.1.1.1"), metav1.CreateOptions{})
require.NoError(t, err)
_, err = k8s.CoreV1().Services("baz").Create(context.Background(), lbService("foo", "2.2.2.2"), metav1.CreateOptions{})
require.NoError(t, err)
_, err = k8s.CoreV1().Services("bat").Create(context.Background(), lbService("foo", "3.3.3.3"), metav1.CreateOptions{})
require.NoError(t, err)

longRunningChan := runCommandAsynchronously(&cmd, []string{
"-addresses", "127.0.0.1",
"-http-port", strconv.Itoa(testClient.Cfg.HTTPPort),
"-consul-write-interval", "100ms",
"-add-k8s-namespace-suffix",
})
defer stopCommand(t, &cmd, longRunningChan)

// check that the name of the service is namespaced
retry.Run(t, func(r *retry.R) {
svc, _, err := consulClient.Catalog().Service("foo-bar", "k8s", nil)
require.NoError(r, err)
require.Len(r, svc, 1)
require.Equal(r, "1.1.1.1", svc[0].ServiceAddress)
svc, _, err = consulClient.Catalog().Service("foo-baz", "k8s", nil)
require.NoError(r, err)
require.Len(r, svc, 1)
require.Equal(r, "2.2.2.2", svc[0].ServiceAddress)
svc, _, err = consulClient.Catalog().Service("foo-bat", "k8s", nil)
require.NoError(r, err)
require.Len(r, svc, 1)
require.Equal(r, "3.3.3.3", svc[0].ServiceAddress)
})

exitChan := runCommandAsynchronously(&cmd, []string{
"-addresses", "127.0.0.1",
"-http-port", strconv.Itoa(testClient.Cfg.HTTPPort),
"-purge-k8s-services-from-node=k8s-sync",
"-filter=baz in ID",
})
stopCommand(t, &cmd, exitChan)

retry.Run(t, func(r *retry.R) {
serviceList, _, err := consulClient.Catalog().NodeServiceList("k8s-sync", &api.QueryOptions{AllowStale: false})
require.NoError(r, err)
require.Len(r, serviceList.Services, 2)
})
}

// Set up test consul agent and fake kubernetes cluster client.
func completeSetup(t *testing.T) (*fake.Clientset, *test.TestServerClient) {
k8s := fake.NewSimpleClientset()
Expand Down

0 comments on commit 82ed2e7

Please sign in to comment.