Как etcd watch отслеживает удаление сервиса из реестра Service Discovery?

23.04.2024

etcd, Архитектура

Комментариев нет

В etcd, механизм watch отслеживает изменения в реестре данных, включая как добавление, так и удаление данных. Когда вы используете watch с префиксом или конкретным ключом, etcd будет уведомлять вас не только о добавлении (PUT) или изменении значений, но и об удалении (DELETE) данных.

Как watch отслеживает удаление данных?

Когда сервис удаляется из реестра, например, с помощью команды Delete, etcd генерирует событие типа DELETE. Это событие включает информацию о ключе, который был удален.

Таким образом, если вы используете watch для отслеживания ключей, например, services/, то если какой-либо сервис удаляется из etcd, вы получите уведомление о событии DELETE с указанием ключа удаленного сервиса.

Пример работы с удалением данных с использованием watch:

Предположим, что мы зарегистрировали сервис с ключом services/service1, а затем этот сервис был удален из реестра. Мы можем отследить это с помощью watch, который уведомит нас о событии DELETE.

Пример кода, который отслеживает удаление сервиса:

package main

import (
	"context"
	"fmt"
	"log"
	"go.etcd.io/etcd/v3/clientv3"
	"time"
)

func main() {
	// Подключаемся к серверу etcd
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   []string{"localhost:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatalf("Ошибка при подключении к etcd: %v", err)
	}
	defer client.Close()

	// Регистрация сервиса
	registerService(client, "service1", "localhost:8080")

	// Запуск мониторинга сервисов
	go watchServices(client)

	// Ожидаем некоторое время, чтобы наблюдать за событиями
	time.Sleep(5 * time.Second)

	// Удаляем сервис для демонстрации отслеживания события удаления
	deleteService(client, "service1")

	// Ожидаем некоторое время для обработки событий
	time.Sleep(10 * time.Second)
}

// Регистрация сервиса в etcd
func registerService(client *clientv3.Client, serviceName, serviceAddress string) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := "services/" + serviceName
	value := serviceAddress

	// Регистрируем сервис в etcd
	_, err := client.Put(ctx, key, value)
	if err != nil {
		log.Fatalf("Ошибка при регистрации сервиса в etcd: %v", err)
	}
	fmt.Printf("Сервис %s зарегистрирован по адресу %s\n", serviceName, serviceAddress)
}

// Удаление сервиса из etcd
func deleteService(client *clientv3.Client, serviceName string) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	key := "services/" + serviceName

	// Удаляем сервис из etcd
	_, err := client.Delete(ctx, key)
	if err != nil {
		log.Fatalf("Ошибка при удалении сервиса из etcd: %v", err)
	}
	fmt.Printf("Сервис %s удален\n", serviceName)
}

// Мониторинг сервисов в etcd
func watchServices(client *clientv3.Client) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Настроим watch на ключи с префиксом "services/"
	watchChan := client.Watch(ctx, "services/", clientv3.WithPrefix())

	fmt.Println("Запуск мониторинга сервисов...")

	for watchResp := range watchChan {
		for _, ev := range watchResp.Events {
			switch ev.Type {
			case clientv3.EventTypePut:
				fmt.Printf("Новый сервис зарегистрирован: %s = %s\n", ev.Kv.Key, ev.Kv.Value)
			case clientv3.EventTypeDelete:
				fmt.Printf("Сервис удален: %s\n", ev.Kv.Key)
			}
		}
	}
}

package main

import (

"context"

"fmt"

"log"

"go.etcd.io/etcd/v3/clientv3"

"time"

)

func main() {

// Подключаемся к серверу etcd

client, err := clientv3.New(clientv3.Config{

Endpoints: []string{"localhost:2379"},

DialTimeout: 5 * time.Second,

})

if err != nil {

log.Fatalf("Ошибка при подключении к etcd: %v", err)

}

defer client.Close()

// Регистрация сервиса

registerService(client, "service1", "localhost:8080")

// Запуск мониторинга сервисов

go watchServices(client)

// Ожидаем некоторое время, чтобы наблюдать за событиями

time.Sleep(5 * time.Second)

// Удаляем сервис для демонстрации отслеживания события удаления

deleteService(client, "service1")

// Ожидаем некоторое время для обработки событий

time.Sleep(10 * time.Second)

}

// Регистрация сервиса в etcd

func registerService(client *clientv3.Client, serviceName, serviceAddress string) {

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

key := "services/" + serviceName

value := serviceAddress

// Регистрируем сервис в etcd

_, err := client.Put(ctx, key, value)

if err != nil {

log.Fatalf("Ошибка при регистрации сервиса в etcd: %v", err)

}

fmt.Printf("Сервис %s зарегистрирован по адресу %s\n", serviceName, serviceAddress)

}

// Удаление сервиса из etcd

func deleteService(client *clientv3.Client, serviceName string) {

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

key := "services/" + serviceName

// Удаляем сервис из etcd

_, err := client.Delete(ctx, key)

if err != nil {

log.Fatalf("Ошибка при удалении сервиса из etcd: %v", err)

}

fmt.Printf("Сервис %s удален\n", serviceName)

}

// Мониторинг сервисов в etcd

func watchServices(client *clientv3.Client) {

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

defer cancel()

// Настроим watch на ключи с префиксом "services/"

watchChan := client.Watch(ctx, "services/", clientv3.WithPrefix())

fmt.Println("Запуск мониторинга сервисов...")

for watchResp := range watchChan {

for _, ev := range watchResp.Events {

switch ev.Type {

case clientv3.EventTypePut:

fmt.Printf("Новый сервис зарегистрирован: %s = %s\n", ev.Kv.Key, ev.Kv.Value)

case clientv3.EventTypeDelete:

fmt.Printf("Сервис удален: %s\n", ev.Kv.Key)

}

Объяснение:

registerService — Регистрирует сервис в etcd с ключом services/service1 и значением localhost:8080.
deleteService — Удаляет сервис из etcd с тем же ключом services/service1, что имитирует удаление сервиса.
watchServices — Использует watch для мониторинга изменений в префиксе services/. Когда сервис регистрируется (событие PUT), мы выводим сообщение о добавлении. Когда сервис удаляется (событие DELETE), выводится сообщение о его удалении.

Как это работает:

Когда сервис регистрируется, etcd генерирует событие PUT, и вы видите, что сервис добавлен.
Когда сервис удаляется из etcd с помощью Delete, etcd генерирует событие DELETE, и вы видите сообщение, что сервис был удален.

Вывод:

Сервис service1 зарегистрирован по адресу localhost:8080
Запуск мониторинга сервисов...
Новый сервис зарегистрирован: services/service1 = localhost:8080
Сервис service1 удален
Сервис service1 удален

Сервис service1 зарегистрирован по адресу localhost:8080

Запуск мониторинга сервисов...

Новый сервис зарегистрирован: services/service1 = localhost:8080

Сервис service1 удален

Важные моменты:

Если вы хотите отслеживать только удаление, вы можете просто обработать событие типа DELETE.
watch поддерживает отслеживание изменений с префиксом, что полезно для мониторинга всех сервисов, зарегистрированных под ключами с одинаковым префиксом, как в примере с services/.

Таким образом, использование watch позволяет не только отслеживать добавление или изменение сервисов, но и своевременно реагировать на их удаление из реестра.

Как etcd watch отслеживает удаление сервиса из реестра Service Discovery?

23.04.2024

etcd, Архитектура

Комментариев нет

Как watch отслеживает удаление данных?

Пример работы с удалением данных с использованием watch:

Пример кода, который отслеживает удаление сервиса:

Объяснение:

Как это работает:

Вывод:

Важные моменты:

Recommended Posts

Golang: компонент регистрации в etcd (Service Discovery)

etcd watch и реализации механизма обнаружения сервисов (Service Discovery) в микросервисной архитектуре

health-check API для микросервисов

Добавить комментарий Отменить ответ