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

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

StartLiveTail与 Amazon SDK 或 CLI 配合使用

以下代码示例演示如何使用 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
适用于 Go V2 的 SDK

包含所需的文件。

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 的详细信息,请参阅 Amazon SDK for Go API 参考StartLiveTail中的。

Java
适用于 Java 2.x 的 SDK

包含所需的文件。

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
适用于 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
适用于 Kotlin 的 SDK

包含所需的文件。

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于 K otlin 的Amazon SDK API 参考

Python
SDK for 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 的详细信息,请参阅适用StartLiveTailPython 的Amazon SDK (Boto3) API 参考

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