Использование Temporal для создания масштабируемых и отказоустойчивых приложений в Golang

мы расскажем, как создавать готовые к запуску сервисы с помощью Temporal в Golang. В этом учебном пособии вы узнаете, как проектировать и создавать отказоустойчивые и масштабируемые сервисы, подходящие для использования в распределенных системах.

Шаг 1. Запустите Temporal локально с помощью Docker

Если вы хотите запускать Temporal локально для тестирования и разработки, вы можете использовать Docker. Команда Temporal предоставляет официальный образ Docker, который можно использовать для локального запуска всего стека Temporal.

Для начала убедитесь, что на вашем компьютере установлен Docker. Скачать Docker можно с официального сайта.

После того, как вы установили Docker, вы можете использовать официальную настройку docker-compose, предоставленную Temporal, для локального запуска всего стека Temporal с помощью нескольких простых команд.

Для начала выполните временный Быстрый старт для разработки на локальном хосте или загрузите установочные файлы docker-compose, выполнив следующую команду:

curl -o docker-compose.yml https://raw.githubusercontent.com/temporalio/docker-compose/master/docker-compose-cas.yml

Это загрузит установку docker-compose для Temporal Community Edition с хранилищем данных Cassandra.

Затем запустите службы, выполнив следующую команду:

docker-compose up

Эта команда запустит необходимые службы, включая сервер Temporal, базу данных Cassandra и веб-интерфейс Temporal для мониторинга рабочих процессов. После запуска служб вы можете использовать Temporal CLI для взаимодействия с сервером Temporal и веб-интерфейс Temporal на http://localhost:8080 для мониторинга ваших рабочих процессов.

docker-compose exec temporal-server tctl namespace list

Чтобы остановить службы, выполните следующую команду:

docker-compose down

Это закроет и удалит все контейнеры, определенные в файле docker-compose.yml.

Локальный запуск Temporal с помощью Docker — отличный способ разработать и протестировать рабочие процессы перед их развертыванием в рабочей среде.

Шаг 2. Настройте рабочее пространство

Создайте новый проект в Golang и импортируйте Temporal SDK, используя следующие команды:

Инициальный мод

go mod init goenv // you can change your project name goenv -> your project name
go get go.temporal.io/sdk

Шаг 3. Разработайте рабочий процесс

Определите долгосрочный бизнес-процесс, который вы хотите организовать, используя код Golang. Используйте Temporal API для моделирования вашего рабочего процесса, обеспечивая правильную обработку исключений и сбоев. Вот пример рабочего процесса для получения и обработки данных о погоде:

package workflow

import (
 "goenv/activity"
 "goenv/messages"
 "time"

 "go.temporal.io/sdk/workflow"
)

// define the workflow function
func WeatherWorkflow(ctx workflow.Context, cityName string) ([]messages.WeatherData, error) {
 options := workflow.ActivityOptions{
  StartToCloseTimeout: time.Second * 5,
 }
 ctx = workflow.WithActivityOptions(ctx, options)

 // start the activities
 currentWeatherFuture := workflow.ExecuteActivity(ctx, activity.GetWeather, cityName)

 // wait for activities to complete
 var current messages.WeatherData
 if err := currentWeatherFuture.Get(ctx, &current); err != nil {
  return nil, err
 }

 var response []messages.WeatherData
 // combine results
 response = append(response, current)

 return response, nil
}

Шаг 4. Определите свои действия

Определите отдельные задачи, составляющие ваш рабочий процесс, с помощью Temporal API. Эти действия должны быть инкапсулированы в автономные функции, которые должны быть идемпотентными и повторяемыми. Вот как определить действие для получения данных о погоде:

// activity/main.go

package activity

import (
 "context"
 "goenv/messages"
 "goenv/store"
)

func GetWeather(ctx context.Context, cityName string) (result messages.WeatherData, err error) {
 result, err = store.GetCurrentWeather(ctx, cityName)
 if err != nil {
  return result, err
 }
 return result, nil
}

Шаг 5. Реализуйте свой рабочий процесс и действия

Напишите код Golang, чтобы создать свой рабочий процесс и действия, используя Temporal API, чтобы обеспечить правильное выполнение и обработку исключений. Вот как реализовать пример рабочего процесса:

// workflow/main.go

package workflow

import (
 "goenv/activity"
 "log"

 "go.temporal.io/sdk/client"
 "go.temporal.io/sdk/worker"
)

func main() {
 c, err := client.Dial(client.Options{})
 if err != nil {
  log.Fatalln("unable to create Temporal client", err)
 }
 defer c.Close()

 w := worker.New(c, "weather", worker.Options{})
 w.RegisterWorkflow(WeatherWorkflow)
 w.RegisterActivity(activity.GetWeather)
 // Start listening to the Task Queue
 err = w.Run(worker.InterruptCh())
 if err != nil {
  log.Fatalln("unable to start Worker", err)
 }
}

Шаг 6. Обработка невыполненных задач

В распределенной системе ошибки неизбежны. Одна из самых важных вещей, о которой следует помнить при создании системы с помощью Temporal, — это правильная обработка ошибок. У вас должен быть план на тот случай, если что-то пойдет не так, например, при сетевом подключении или ошибках ввода данных пользователем.

Когда задача завершается сбоем и возвращается ошибка, есть два подхода: повторить задачу или проигнорировать ошибку. Выбранный вами подход зависит от типа ошибки и степени важности задачи для рабочего процесса.

Чтобы повторить невыполненную задачу, вы можете использовать метод NewRetriableApplicationError Temporal SDK. NewRetriableApplicationError принимает три параметра: строку сообщения, токен задачи и ошибку. Маркер задачи — это уникальная подстрока, используемая для идентификации конкретной задачи.

Вот пример:

// activity/main.go

package activity

import (
 "context"
 "goenv/messages"
 "goenv/store"

 "go.temporal.io/sdk/temporal"
)

func GetWeather(ctx context.Context, cityName string) (result messages.WeatherData, err error) {
 result, err = store.GetCurrentWeather(ctx, cityName)
 if err != nil {
  return result, temporal.NewApplicationError("unable to get weather data", "GET_WEATHER", err)
 }
 return result, nil
}

В этом примере при сбое вызова GetWeather задача помечается для повторной попытки с помощью маркера задачи retry-getweather и возвращается ошибка.

С другой стороны, иногда ошибки не важны для рабочего процесса и могут быть проигнорированы. В этом случае вы можете вернуть ненулевую ошибку из функции действия, и эта ошибка будет зарегистрирована, но рабочий процесс продолжится. Например, если GetWeather не может найти данные о погоде для указанного города, и эти данные не являются критическими для рабочего процесса, вы можете просто вернуть ненулевую ошибку и продолжить рабочий процесс.

func GetWeather(ctx context.Context, cityName string, result chan<- messages.WeatherData) error {
    weather, err := store.GetWeather(cityName, weatherType)
    if err != nil {
        logger.Warnf(ctx, "failed to get weather data: %v", err)
        return nil
    }
    result <- weather
    return nil
}

В этом примере, если GetWeather возвращает ошибку, она регистрируется в журнале, а действие возвращает нулевую ошибку. Это приведет к тому, что рабочий процесс продолжит работу без этих конкретных данных о погоде.

Обязательно обрабатывайте ошибки надлежащим образом и регулярно тестируйте приложение, чтобы убедиться, что оно устойчиво к сбоям.

Шаг 7. Проверьте свою систему

Используйте Temporal SDK для тестирования вашей системы, моделирования сбоев и проверки отказоустойчивости. Вы можете использовать утилиту TemporalTestSuite, входящую в состав Temporal SDK, для тестирования вашей системы. Вот пример теста:

// workflow/test/workflow_test.go

package test

import (
 "goenv/activity"
 "goenv/messages"
 "goenv/workflow"
 "testing"

 "github.com/stretchr/testify/mock"
 "github.com/stretchr/testify/require"
 "go.temporal.io/sdk/testsuite"
)

func TestWeatherWorkflow(t *testing.T) {
 // Set up the test suite and testing execution environment
 testSuite := &testsuite.WorkflowTestSuite{}
 env := testSuite.NewTestWorkflowEnvironment()

 // Mock activity implementation
 env.OnActivity(activity.GetWeather, mock.Anything, mock.Anything).Return(messages.WeatherData{
  Temperature: 41,
  Humidity:    80,
  WindSpeed:   4,
 }, nil)

 env.ExecuteWorkflow(workflow.WeatherWorkflow, "Cairo")

 require.True(t, env.IsWorkflowCompleted())
 require.NoError(t, env.GetWorkflowError())

 var data []messages.WeatherData
 require.NoError(t, env.GetWorkflowResult(&data))
 require.Equal(t, []messages.WeatherData{
  {
   Temperature: 41,
   Humidity:    80,
   WindSpeed:   4,
  },
 }, data)
}

Этот тест выполняет рабочий процесс WeatherWorkflow с заданным cityName и использует TemporalOption для настройки TaskQueue для действия GetWeather.

Шаг 8. Используйте Temporal Cloud и настройте сервер REST

Чтобы воспользоваться всеми преимуществами Temporal, вы можете использовать Temporal Cloud — полностью управляемую службу, обеспечивающую масштабируемую и надежную оркестровку ваших рабочих процессов. Temporal Cloud занимается развертыванием, масштабированием и обслуживанием Temporal, поэтому вы можете сосредоточиться на создании своего приложения.

Чтобы начать работу с Temporal Cloud, создайте учетную запись и получите необходимые учетные данные. Вам нужно настроить Temporal SDK, чтобы использовать эти учетные данные.

Затем вы можете настроить сервер Golang Mux, чтобы ваши рабочие процессы отображались как службы REST. Mux — популярная библиотека HTTP-маршрутизации для Golang. Вот пример того, как настроить простой сервер:

// main.go
package main

import (
 "goenv/activity"
 "goenv/handler"
 "goenv/workflow"
 "log"
 "net/http"

 "go.temporal.io/sdk/client"
 "go.temporal.io/sdk/worker"
)

func main() {
 // set up the worker
 c, err := client.Dial(client.Options{})
 if err != nil {
  log.Fatalln("unable to create Temporal client", err)
 }
 defer c.Close()

 w := worker.New(c, "weather", worker.Options{})
 w.RegisterWorkflow(workflow.WeatherWorkflow)
 w.RegisterActivity(activity.GetWeather)

 mux := http.NewServeMux()
 mux.HandleFunc("/weather", handler.WeatherHandler) // curl -X GET http://localhost:8080/weather?city=Cairo
 server := &http.Server{Addr: ":5000", Handler: mux}

 // start the worker and the web server
 go w.Run(worker.InterruptCh())
 log.Fatal(server.ListenAndServe())
}
// handler/weather.go

package handler

import (
 "encoding/json"
 "goenv/messages"
 "goenv/workflow"
 "log"
 "net/http"

 "go.temporal.io/sdk/client"
)

func WeatherHandler(w http.ResponseWriter, r *http.Request) {
 // execute weather workflow with the city name from request query
 cityName := r.URL.Query().Get("city")
 if cityName == "" {
  http.Error(w, "city name is required", http.StatusBadRequest)
  return
 }

 // create a new temporal client
 // set up the worker
 c, err := client.Dial(client.Options{})
 if err != nil {
  log.Fatalln("unable to create Temporal client", err)
 }
 defer c.Close()

 we, err := c.ExecuteWorkflow(r.Context(), client.StartWorkflowOptions{
  ID:        "weather_workflow",
  TaskQueue: "weather",
 }, workflow.WeatherWorkflow, cityName)
 if err != nil {
  http.Error(w, "unable to start workflow", http.StatusInternalServerError)
  return
 }

 // wait for workflow to complete
 var result []messages.WeatherData
 if err := we.Get(r.Context(), &result); err != nil {
  http.Error(w, "unable to get workflow result", http.StatusInternalServerError)
  return
 }

 // convert result to json in key-value pais
 response := make(map[string]interface{})
 for _, data := range result {
  response[cityName] = data
 }

 jsonResponse, err := json.Marshal(response)
 if err != nil {
  http.Error(w, "unable to marshal response", http.StatusInternalServerError)
  return
 }

 w.Header().Set("Content-Type", "application/json")
 w.Write(jsonResponse)
}

Функция WeatherHandler берет название города из входящего HTTP-запроса и запускает рабочий процесс WeatherWorkflow с этим вводом, используя функцию Temporal ExecuteWorkflow. Затем он ждет результата и возвращает его клиенту.

Теперь вы можете запустить свой http-сервер

go run main.go

и выполните пример рабочего процесса

curl -X GET http://localhost:8080/weather?city=Cairo

и это приведет к успешному временному рабочему процессу, который вернет ответ json

// Example JSON Response

{
  "Cairo": {
    Temperature: 36,
    Humidity:    40,
    WindSpeed:   21,
  } 
}

Рабочий процесс во временной панели мониторинга

Пример кода: https://github.com/unijad/temporal-tutorial-example-01