StartLiveTail搭配使用 Amazon SDK - Amazon CloudWatch 日志
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

StartLiveTail搭配使用 Amazon SDK

以下代码示例演示如何使用 StartLiveTail

.NET
Amazon SDK for .NET

包含所需的文件。

using Amazon; using Amazon.CloudWatchLogs; using Amazon.CloudWatchLogs.Model;

启动 Live Tail 会话

var client = new AmazonCloudWatchLogsClient(); var request = new StartLiveTailRequest { LogGroupIdentifiers = logGroupIdentifiers, LogStreamNames = logStreamNames, LogEventFilterPattern = filterPattern, }; var response = await client.StartLiveTailAsync(request); // Catch if request fails if (response.HttpStatusCode != System.Net.HttpStatusCode.OK) { Console.WriteLine("Failed to start live tail session"); return; }

您可以通过两种方式处理 Live Tail 会话中的事件:

/* Method 1 * 1). Asynchronously loop through the event stream * 2). Set a timer to dispose the stream and stop the Live Tail session at the end. */ var eventStream = response.ResponseStream; var task = Task.Run(() => { foreach (var item in eventStream) { if (item is LiveTailSessionUpdate liveTailSessionUpdate) { foreach (var sessionResult in liveTailSessionUpdate.SessionResults) { Console.WriteLine("Message : {0}", sessionResult.Message); } } if (item is LiveTailSessionStart) { Console.WriteLine("Live Tail session started"); } // On-stream exceptions are processed here if (item is CloudWatchLogsEventStreamException) { Console.WriteLine($"ERROR: {item}"); } } }); // Close the stream to stop the session after a timeout if (!task.Wait(TimeSpan.FromSeconds(10))){ eventStream.Dispose(); Console.WriteLine("End of line"); }
/* Method 2 * 1). Add event handlers to each event variable * 2). Start processing the stream and wait for a timeout using AutoResetEvent */ AutoResetEvent endEvent = new AutoResetEvent(false); var eventStream = response.ResponseStream; using (eventStream) // automatically disposes the stream to stop the session after execution finishes { eventStream.SessionStartReceived += (sender, e) => { Console.WriteLine("LiveTail session started"); }; eventStream.SessionUpdateReceived += (sender, e) => { foreach (LiveTailSessionLogEvent logEvent in e.EventStreamEvent.SessionResults){ Console.WriteLine("Message: {0}", logEvent.Message); } }; // On-stream exceptions are captured here eventStream.ExceptionReceived += (sender, e) => { Console.WriteLine($"ERROR: {e.EventStreamException.Message}"); }; eventStream.StartProcessing(); // Stream events for this amount of time. endEvent.WaitOne(TimeSpan.FromSeconds(10)); Console.WriteLine("End of line"); }
  • 有关API详细信息,请参阅 “Amazon SDK for .NET API参考 StartLiveTail” 中的。

Go
SDK适用于 Go V2

包含所需的文件。

import ( "context" "log" "time" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs" "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types" )

处理 Live Tail 会话中的事件。

func handleEventStreamAsync(stream *cloudwatchlogs.StartLiveTailEventStream) { eventsChan := stream.Events() for { event := <-eventsChan switch e := event.(type) { case *types.StartLiveTailResponseStreamMemberSessionStart: log.Println("Received SessionStart event") case *types.StartLiveTailResponseStreamMemberSessionUpdate: for _, logEvent := range e.Value.SessionResults { log.Println(*logEvent.Message) } default: // Handle on-stream exceptions if err := stream.Err(); err != nil { log.Fatalf("Error occured during streaming: %v", err) } else if event == nil { log.Println("Stream is Closed") return } else { log.Fatalf("Unknown event type: %T", e) } } } }

启动 Live Tail 会话。

cfg, err := config.LoadDefaultConfig(context.TODO()) if err != nil { panic("configuration error, " + err.Error()) } client := cloudwatchlogs.NewFromConfig(cfg) request := &cloudwatchlogs.StartLiveTailInput{ LogGroupIdentifiers: logGroupIdentifiers, LogStreamNames: logStreamNames, LogEventFilterPattern: logEventFilterPattern, } response, err := client.StartLiveTail(context.TODO(), request) // Handle pre-stream Exceptions if err != nil { log.Fatalf("Failed to start streaming: %v", err) } // Start a Goroutine to handle events over stream stream := response.GetStream() go handleEventStreamAsync(stream)

经过一段时间后停止 Live Tail 会话。

// Close the stream (which ends the session) after a timeout time.Sleep(10 * time.Second) stream.Close() log.Println("Event stream closed")
  • 有关API详细信息,请参阅 “适用于 Go 的 Amazon SDK API参考 StartLiveTail” 中的。

Java
SDK适用于 Java 2.x

包含所需的文件。

import io.reactivex.FlowableSubscriber; import io.reactivex.annotations.NonNull; import org.reactivestreams.Subscription; import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider; import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsAsyncClient; import software.amazon.awssdk.services.cloudwatchlogs.model.LiveTailSessionLogEvent; import software.amazon.awssdk.services.cloudwatchlogs.model.LiveTailSessionStart; import software.amazon.awssdk.services.cloudwatchlogs.model.LiveTailSessionUpdate; import software.amazon.awssdk.services.cloudwatchlogs.model.StartLiveTailRequest; import software.amazon.awssdk.services.cloudwatchlogs.model.StartLiveTailResponseHandler; import software.amazon.awssdk.services.cloudwatchlogs.model.CloudWatchLogsException; import software.amazon.awssdk.services.cloudwatchlogs.model.StartLiveTailResponseStream; import java.util.Date; import java.util.List; import java.util.concurrent.atomic.AtomicReference;

处理 Live Tail 会话中的事件。

private static StartLiveTailResponseHandler getStartLiveTailResponseStreamHandler( AtomicReference<Subscription> subscriptionAtomicReference) { return StartLiveTailResponseHandler.builder() .onResponse(r -> System.out.println("Received initial response")) .onError(throwable -> { CloudWatchLogsException e = (CloudWatchLogsException) throwable.getCause(); System.err.println(e.awsErrorDetails().errorMessage()); System.exit(1); }) .subscriber(() -> new FlowableSubscriber<>() { @Override public void onSubscribe(@NonNull Subscription s) { subscriptionAtomicReference.set(s); s.request(Long.MAX_VALUE); } @Override public void onNext(StartLiveTailResponseStream event) { if (event instanceof LiveTailSessionStart) { LiveTailSessionStart sessionStart = (LiveTailSessionStart) event; System.out.println(sessionStart); } else if (event instanceof LiveTailSessionUpdate) { LiveTailSessionUpdate sessionUpdate = (LiveTailSessionUpdate) event; List<LiveTailSessionLogEvent> logEvents = sessionUpdate.sessionResults(); logEvents.forEach(e -> { long timestamp = e.timestamp(); Date date = new Date(timestamp); System.out.println("[" + date + "] " + e.message()); }); } else { throw CloudWatchLogsException.builder().message("Unknown event type").build(); } } @Override public void onError(Throwable throwable) { System.out.println(throwable.getMessage()); System.exit(1); } @Override public void onComplete() { System.out.println("Completed Streaming Session"); } }) .build(); }

启动 Live Tail 会话。

CloudWatchLogsAsyncClient cloudWatchLogsAsyncClient = CloudWatchLogsAsyncClient.builder() .credentialsProvider(ProfileCredentialsProvider.create()) .build(); StartLiveTailRequest request = StartLiveTailRequest.builder() .logGroupIdentifiers(logGroupIdentifiers) .logStreamNames(logStreamNames) .logEventFilterPattern(logEventFilterPattern) .build(); /* Create a reference to store the subscription */ final AtomicReference<Subscription> subscriptionAtomicReference = new AtomicReference<>(null); cloudWatchLogsAsyncClient.startLiveTail(request, getStartLiveTailResponseStreamHandler(subscriptionAtomicReference));

经过一段时间后停止 Live Tail 会话。

/* Set a timeout for the session and cancel the subscription. This will: * 1). Close the stream * 2). Stop the Live Tail session */ try { Thread.sleep(10000); } catch (InterruptedException e) { throw new RuntimeException(e); } if (subscriptionAtomicReference.get() != null) { subscriptionAtomicReference.get().cancel(); System.out.println("Subscription to stream closed"); }
  • 有关API详细信息,请参阅 “Amazon SDK for Java 2.x API参考 StartLiveTail” 中的。

JavaScript
SDK对于 JavaScript (v3)

包含所需的文件。

import { CloudWatchLogsClient, StartLiveTailCommand } from "@aws-sdk/client-cloudwatch-logs";

处理 Live Tail 会话中的事件。

async function handleResponseAsync(response) { try { for await (const event of response.responseStream) { if (event.sessionStart !== undefined) { console.log(event.sessionStart); } else if (event.sessionUpdate !== undefined) { for (const logEvent of event.sessionUpdate.sessionResults) { const timestamp = logEvent.timestamp; const date = new Date(timestamp); console.log("[" + date + "] " + logEvent.message); } } else { console.error("Unknown event type"); } } } catch (err) { // On-stream exceptions are captured here console.error(err) } }

启动 Live Tail 会话。

const client = new CloudWatchLogsClient(); const command = new StartLiveTailCommand({ logGroupIdentifiers: logGroupIdentifiers, logStreamNames: logStreamNames, logEventFilterPattern: filterPattern }); try{ const response = await client.send(command); handleResponseAsync(response); } catch (err){ // Pre-stream exceptions are captured here console.log(err); }

经过一段时间后停止 Live Tail 会话。

/* Set a timeout to close the client. This will stop the Live Tail session. */ setTimeout(function() { console.log("Client timeout"); client.destroy(); }, 10000);
  • 有关API详细信息,请参阅 “Amazon SDK for JavaScript API参考 StartLiveTail” 中的。

Kotlin
SDK对于 Kotlin 来说

包含所需的文件。

import aws.sdk.kotlin.services.cloudwatchlogs.CloudWatchLogsClient import aws.sdk.kotlin.services.cloudwatchlogs.model.StartLiveTailRequest import aws.sdk.kotlin.services.cloudwatchlogs.model.StartLiveTailResponseStream import kotlinx.coroutines.flow.takeWhile

启动 Live Tail 会话

val client = CloudWatchLogsClient.fromEnvironment() val request = StartLiveTailRequest { logGroupIdentifiers = logGroupIdentifiersVal logStreamNames = logStreamNamesVal logEventFilterPattern = logEventFilterPatternVal } val startTime = System.currentTimeMillis() try { client.startLiveTail(request) { response -> val stream = response.responseStream if (stream != null) { /* Set a timeout to unsubcribe from the flow. This will: * 1). Close the stream * 2). Stop the Live Tail session */ stream.takeWhile { System.currentTimeMillis() - startTime < 10000 }.collect { value -> if (value is StartLiveTailResponseStream.SessionStart) { println(value.asSessionStart()) } else if (value is StartLiveTailResponseStream.SessionUpdate) { for (e in value.asSessionUpdate().sessionResults!!) { println(e) } } else { throw IllegalArgumentException("Unknown event type") } } } else { throw IllegalArgumentException("No response stream") } } } catch (e: Exception) { println("Exception occurred during StartLiveTail: $e") System.exit(1) }
  • 有关API详细信息,请参阅StartLiveTail中的 Kotlin Amazon SDK API 参考

Python
SDK适用于 Python (Boto3)

包含所需的文件。

import boto3 import time from datetime import datetime

启动 Live Tail 会话

# Initialize the client client = boto3.client('logs') start_time = time.time() try: response = client.start_live_tail( logGroupIdentifiers=log_group_identifiers, logStreamNames=log_streams, logEventFilterPattern=filter_pattern ) event_stream = response['responseStream'] # Handle the events streamed back in the response for event in event_stream: # Set a timeout to close the stream. # This will end the Live Tail session. if (time.time() - start_time >= 10): event_stream.close() break # Handle when session is started if 'sessionStart' in event: session_start_event = event['sessionStart'] print(session_start_event) # Handle when log event is given in a session update elif 'sessionUpdate' in event: log_events = event['sessionUpdate']['sessionResults'] for log_event in log_events: print('[{date}] {log}'.format(date=datetime.fromtimestamp(log_event['timestamp']/1000),log=log_event['message'])) else: # On-stream exceptions are captured here raise RuntimeError(str(event)) except Exception as e: print(e)
  • 有关API详细信息,请参阅StartLiveTail中的 Amazon SDKPython (Boto3) API 参考。

有关 Amazon SDK开发者指南和代码示例的完整列表,请参阅在 Amazon SDK 中使用 CloudWatch 日志。本主题还包括有关入门的信息以及有关先前SDK版本的详细信息。