Niedawno pisałem już o Step functions. Dla mnie są one czymś w rodzaju języka programowania. Za ich pomocą możemy tworzyć skomplikowane przepływy (workflows) korzystające z różnych komponentów AWS.

Jestem teraz w trakcie realizacji projektu, który wykorzystuje między innymi te maszyny stanów, mam więc okazję, żeby pokazać kilka przykładów łącznie z kodem. Nie będzie niestety rzeczy z życia wziętych (tajne przez poufne), ale postaram się pokazać coś dla każdego ze stanów.

Jak wiemy mamy pięć różnych stanów:

• Task – umozliwia wykonanie czynności,
• Choice – decyzja pomiędzy wieloma czynnościami,
• Wait – zapewnia opóźnienie pomiędzy wykonaniami,
• Parallel – uruchamia równolegle kilka czynności,
• Pass – przechodzi do następnego stanu,
• Succeed / Fail – zatrzymuje wykonanie ze statusem Fail  lub Success.

O Task pisałem poprzednio. Dziś zajmiemy się pozostałymi rodzajami stanów. Przy okazji poprogramujemy trochę.

Choice – zamiast ifów

Mamy jakąś monolityczną aplikację uruchamianą na maszynie wirtualnej lub, w lepszym przypadku, dużą Lambdę z kilkoma ifami w kodzie. W zależności od dostarczonych danych, aplikacja ma wykonać różne czynności. W takim przypadku wykorzystamy typ Choice. 

Zakładam, że mamy cały czas dostępną Lambdę CombineStrings. Jeżeli nie, wracamy do poprzedniego wpisu o Step functions i tworzymy ją. Potrzebna będzie nam też druga Lambda, która poza łączeniem stringów będzie zamieniała wszystkie litery na duże. Kod w Pythonie jest bardzo prosty:

def lambda_handler(event, context):
    strings = ' '.join(event['strings']).upper()
    return strings

Nazwijmy tą funkcję ToUpper i wracamy do Step functions.

Założenie jest takie, że w zależności od wartości zmiennej foo przekazanej do naszej maszynki, stringi przesłane w tablicy strings są ze sobą tylko łączone, czy także wszystkie litery zamieniane są także na duże. Decyzję podejmiemy na podstawie wartości magicznej zmiennej foo. Jeżeli foo ustawimy na 1, wywołamy Lambdę, która połączy wyrazy w jeden string. Jeżeli wartość foo będzie inna (default), oprócz połączenia, zamienimy wszystkie litery na duże. Czyli na wejściu mamy dane

{
    "strings": ["Jakieś", "wyrazy", "do", "połączenia"],
    "foo" : 1 - łączymy | inna wartość - łączymy i zamieniamy litery na duże
 }

i w zależności od wartości przekazanej w zmiennej foo łączymy słowa w jedno zdanie lub także zamieniamy wszystkie litery na duże. Schemat naszej funkcji pokazałem na rynunku poniżej

 

a kod potrzebny do jego stworzenia wygląda tak:

{
  "Comment": "Step function created for fun",
  "StartAt": "TestValue",
  "States": {
    "TestValue": {
      "Type" : "Choice",
      "Choices": [
        {
          "Variable": "$.foo",
          "NumericEquals": 1,
          "Next": "JoinStrings"
        }
      ],
      "Default": "ToUpper"
    },
    "JoinStrings": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:Region:AccountId:function:CombineStrings",
      "ResultPath" : "$.CombinedStrings",
      "End" : true
    },
    "ToUpper": {
      "Type" : "Task",
      "Resource": "arn:aws:lambda:Region:AccountId:function:ToUpper",
      "ResultPath" : "$.CombinedStrings",
      "End": true
    }
  }
}

Po kolei… W linii 3 mamy informację od czego zaczynamy. Umieszczemy tu jeden ze stanów, które opisujemy od linii 4 do 28. U nas jest to: TestValue. Następnie mamy opisane trzy stany: TestValue, JoinStrings i ToUpper. Cała magia dzieje się w stanie TestValue, który sprawdza wartość przekazanej zmiennej foo i na tej podstawie przekazuje wykonanie do stanu JoinStrings lub ToUpper. A te stany to po prostu wywołują odpowiednie funkcje Lambda.

Mamy gotową Step Function, zabieramy się za działanie. W pierwszym wywołaniu przesyłamy dane pokazane poniżej.

{
    "strings": ["Insert", "your", "JSON", "here"],
    "foo" : 1
 }

Wartość zmiennej foo ustawiamy na 1, oczekujemy więc, że wykona się Lambda o nazwie JoinStrings. Chwila niepewności i… sukces. W konsoli możemy sprawdzić jak wykonała się nasza funkcja

i jakie wyniki zwróciła.

Wszystko jest w jak najlepszym porządku. Spróbujmy wysłać coś innego. Tym razem zmienna foo ustawiona jest na 2, czyli nasza step function powinna wywołać Lambdę ToUpper.

{
    "strings": ["Insert", "your", "JSON", "here"],
    "foo" : 2
}

i widzimy, że wszystko fajnie działa.

 

Wywołujemy Step Function z aplikacji

Zanim przejdziemy do następnego stanu, pokażę jak wykorzystać taką funkcję w .NET Core.

Aby skorzystać z dobrodziejstw naszej maszyny stanów musimy dodać paczkę AWSSDK.StepFunctions. Jak zawsze, potrzebujemy klienta.

private static AmazonStepFunctionsClient getClient()
{
    var client = new AmazonStepFunctionsClient(".accessKeyId.", ".secretKey.", Amazon.RegionEndpoint.EUWest1);
    return client;
}

Aby wykonać naszą funkcję potrzebna będzie nam instancja klasy StartExecutionRequest. Za pomocą tego obiektu ustawimy arn do naszej funkcji (StateMachineArn) i przekażemy dane do funkcji (Input). Aby wystartować naszą Step Function wywołujemy metodę klienta StartExecution lub StartExecutionAsync, do których przekazujemy nasz request.

public static async Task<StartExecutionResponse> InvokeStepFunction(string stepFunctionArn, string functionInput)
{
    var client = getClient();
    var request = new StartExecutionRequest();
    if (!string.IsNullOrEmpty(functionInput))
    {
        request.Input = functionInput; 
    }
    request.StateMachineArn = stepFunctionArn;
    var startExecutionResponse = await client.StartExecutionAsync(request);
    return startExecutionResponse;
}

Metoda po wywołaniu powinna zwrócić nam obiekt klasy StartExecutionResponse.

Dobrze jest sprawdzić czy wywołanie przebiegło pomyślnie. HttpStatusCode powinno być ustawione na 200 (OK). W przypadku długo działających funkcji bardzo ważny jest też adres arn do naszego wykonania. Za pomocą tego adresu i obiektu klasy DescribeExecutionRequest możemy sprawdzić stan wywołanej funkcji.

Zakładając, że jako dane wejściowe do naszego wywołania przesłaliśmy

private static readonly string input = "{\"strings\": [\"Insert\", \"your\", \"JSON\", \"here\"],\"foo\" : 1}";

wywołanie metody DescribeExecution lub DescribeExecutionAsync

public static async Task<DescribeExecutionResponse> DescribeExecution(string executionArn)
{
    var client = getClient();
    var describeExecutionRequest = new DescribeExecutionRequest();
    describeExecutionRequest.ExecutionArn = executionArn;
    var describeExecutionResponse = await client.DescribeExecutionAsync(describeExecutionRequest);
    return describeExecutionResponse;
}

powinno zwrócić instancję klasy DescribeExecutionResponse, w której właściwość Output pokazuje nam wynik działania funkcji.

Ważną właściwością w tej klasie jest Status. W naszym przypadku, przed pobraniem wyniku działania funkcji, powinniśmy sprawdzić czy jej wartość nie wskazuje na to, że funkcja nie zakończyła jeszcze działania. Pokażę to w następnym przykładzie dotyczącym stanu Wait.

Sprawdźmy jeszcze, czy przesłanie innej wartości zmiennej foo w danych wejściowych do funkcji zmieni jej wynik.

private static readonly string input = "{\"strings\": [\"Insert\", \"your\", \"JSON\", \"here\"],\"foo\" : 2}";

Timer

Zdarza się czasami, że musimy wstrzymać wykonanie naszej aplikacji na jakiś czas lub do jakiegoś momentu. Możliwość taką zapewnie nam typ stanu Wait. Aby zaprezentować działanie tego stanu w naszym przykładzie użyjemy obu funkcji Lambda, ale przedzielimy je 10-sekundową przerwą.

Kod potrzebny do stworzenia naszej funkcji widać poniżej.

{
  "Comment": "Another step function created for fun",
  "StartAt": "JoinStrings",
  "States": {
    "JoinStrings": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:Region:AccountId:function:CombineStrings",
      "ResultPath" : "$.CombinedStrings",
      "Next" : "WaitAMoment"
    },
    "WaitAMoment" : { 
      "Type": "Wait", 
      "Seconds": 10, 
      "Next": "ToUpper" 
    },
    "ToUpper": {
      "Type" : "Task",
      "Resource": "arn:aws:lambda:Region:AccountId:function:ToUpper",
      "ResultPath" : "$.CombinedStrings",
      "End" : true
    }
  }
}

Dodam tylko, że ten stan może przyjmować dwa rodzaje opóźnienia. W naszym przypadku odczekamy po prostu 10 sekund, ale możliwe jest także ustawienie konkretnej wartości dla czasu wykonania danego kroku. Zakładając, że chcielibyśmy wykonać jakiś krok 31 marca 2018 roku o godzinie 16:00, definicja naszego stanu mogłaby wyglądać tak:

"WaitUntil" : { 
  "Type": "Wait", 
  "Timestamp": "2018-03-31T16:00:00Z", 
  "Next": "ToUpper" 
}

Wartość dla timestamp możemy także przekazać na wejście funkcji i wówczas definiujemy to za pomocą TimestampPath tak:

"WaitUntil" : { 
  "Type": "Wait", 
  "TimestampPath": "$.executionDate", 
  "Next": "ToUpper" 
}

Sprawdzamy stan wykonania funkcji

Stan, czyli ExecutionStatus,  w którym znajduje się nasza funkcja może przybierać jedną z następujących wartości:

• ABORTED
• FAILED
• RUNNING
• SUCCEDED
• TIMED_OUT

W naszym przykładzie mamy 10 sekund opóźnienia przed wykonaniem drugiej Lambdy. Przed pobraniem wartości zwracanej, dobrze jest więc sprawdzić czy funkcja zakończyła działanie.  Sprawdzamy status naszego wywołania. Korzystamy z API  DescribeExecutionRequest  i możemy to zrobić na przykład tak:

public static async Task<string> GetOutputValue(string executionArn)
{
    string result = string.Empty;
    var client = getClient();
    var describeExecutionRequest = new DescribeExecutionRequest();
    describeExecutionRequest.ExecutionArn = executionArn;

    DescribeExecutionResponse describeExecutionResponse;

    do
    {
        try
        {
            describeExecutionResponse = await client.DescribeExecutionAsync(describeExecutionRequest);
            result = describeExecutionResponse?.Output;
            await Task.Delay(1000);
            Console.WriteLine("  Step function status: " + describeExecutionResponse.Status);
        }
        catch (Exception)
        {            
        }
    } while (describeExecutionResponse?.Status == "RUNNING");
    return result;
}

Co jedną sekundę sprawdzamy status wwywołania. W momencie, kiedy funkcja zakończy działanie zwracamy rezultat działania.

Parallel

Ciekawy typ. Umożliwia równoległe wykonanie kilku czynności. Najłatwiej będzie pokazać to na prostym przykładzie. Wywołamy funkcję, a sama funkcja będzie wykonywała równolegle dwie czynności.

Zamiast stanu Task użyjemy typu Pass, który po prostu przekaże wykonywanie dalej. Kod do utworzenia powyższej maszyny stanów wygląda następująco:

{
  "StartAt": "ParallelStart",
  "States": {
    "ParallelStart": {
      "Type": "Parallel",
      "Next": "Final State",
      "Branches": [
        {
          "StartAt": "Wait_1_2",
          "States": {
            "Wait_1_2": {
              "Type": "Wait",
              "Seconds": 20,
              "Next": "Task_1_1"
            },
            "Task_1_1": {
              "Type": "Pass",
              "Result" : "Task 1_1",
              "End": true
            }
          }
        },
        {
          "StartAt": "Task_2_1",
          "States": {
            "Task_2_1": {
              "Type": "Pass",
              "Result" : "Task 2_1",
              "Next": "Wait_2_1"
            },
            "Wait_2_1": {
              "Type": "Wait",
              "Seconds": 20,
              "End": true
            }
          }
        }
      ]
    },
    "Final State": {
      "Type": "Pass",
      "Result" : "Final state",
      "End": true
    }
  }
}

Funkcja wchodzi w stan ParallelStart, w którym rozgałęzia się na dwa „wątki”.

W pierwszym czekamy 20 sekund, a w drugim zwracamy od razu rezultat działania stanu "Result" : "Task 2_1"

i czekamy następnie 20 sekund na zakończenie i przejście do stanu Final State

 

 W tym stanie także zwracamy wynik "Result" : "Final state"

i kończymy działanie funkcji.

Succeed / Fail

Oba typy zatrzymują działanie maszyny stanów. Z różnym skutkiem, sukcesu lub błędu. W ostatnim przykładzie na podstawie wartości zmiennej foo przekazanej do funkcji podejmiemy decyzję co dalej.

Jeżeli wartość zmiennej foo przyjmie wartość 1 zakończymy działanie błędem. Dla wartości 2 zakończymy sukcesem, a dla innych wartości przejdziemy do stanu domyślnego. Czyli pozwolimy naszej maszynie na dalsze działanie.

{
  "StartAt": "Choice",
  "States": {
    "Choice": {
      "Type" : "Choice",
      "Choices": [
        {
          "Variable": "$.foo",
          "NumericEquals": 1,
          "Next": "FailState"
        },
        {
          "Variable": "$.foo",
          "NumericEquals": 2,
          "Next": "SuccessState"
        }
      ],
      "Default": "DefaultState"
    },
    "FailState": {
      "Type" : "Fail",
      "Error" : "Value error",
      "Cause" : "Foo value was set to 1"
    },
    "SuccessState": {
      "Type" : "Succeed"
    },
    "DefaultState" : {
      "Type" : "Pass",
      "Result" : "End",
      "End" : true
    }
  }
}

Zaczniemy od błędu przekażemy do funkcji

{
    "foo" : 1
}

Rezultat działania widać poniżej.

Możemy oczywiście zwrócić na przykład powód błędu

 

Przekazując do funkcji wartość inną niż 1 i 2 pozwolimy na dalsze działanie wywołania. Rezultat działania wygląda następująco.

Dla foo równego 2 maszyna zatrzyma się, ale nie będzie to wykonanie z błędem.

Dlaczego różnicować działanie dla wartości 2 i większych? W rzeczywistości na przykład dalsze działanie dla niektórych wartości nie ma sensu, ale wywołanie nie powinno kończyć się błędem.

Co dalej

Jak wspomniałem, dla mnie Step functions są pewnego rodzaju językiem programowania, za pomocą którego można łączyć działanie wielu usług AWS w większą całość. We wpisie nie pokazałem wszystkich mozliwości. Nie zahaczyliśmy na przykład o obsługę błędów. Warto zainteresować się tematem. W połączeniu z API daje programistom bardzo wiele ciekawych możliwości.