本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 为 Windows 插件创建 Kinesis 代理 对于大多数情况下，不需要为微软 Windows 插件创建 Amazon Kinesis 代理。Windows 的 Kinesis Agent 高度可配置，并包含功能强大的源和接收器，例如`DirectorySource`和`KinesisStream`，这对于大多数情况都足够了。有关现有源和接收器的详细信息，请参阅[为微软 Windows 配置 Amazon Kinesis 运行代理](configuring-kinesis-agent-windows.md)。 对于不常见的场景，可能需要使用定制插件扩展 Windows 的 Kinesis 代理。这样的一些场景包括： + 使用 `Regex` 或 `Delimited` 记录解析程序打包复杂的 `DirectorySource` 声明，使其易于在多种不同类型的配置文件中应用。 + 创建并非基于文件或者超过了现有记录解析程序所提供解析功能的新源。 + 为 AWS 服务创建当前不支持的接收器。 **Topics** + [适用于 Windows 插件的 Kinesis 代理入门](#creating-kinesis-agent-windows-plugins-overview) + [为 Windows 插件工厂实施 Kinesis 代理](#creating-kinesis-agent-windows-plugins-factory) + [实现 Windows 插件 Kinesis 代理程序](#creating-kinesis-agent-windows-plugins-source) + [实现 Windows 插件接收器的 Kinesis 代理](#creating-kinesis-agent-windows-plugins-sink) ## 适用于 Windows 插件的 Kinesis 代理入门 关于自定义插件没有特殊要求。所有现有源和接收器使用的机制，与自定义插件在 Windows 启动时加载的机制相同，在读取`appsettings.json`配置文件。 当 Windows 的 Kinesis 代理程序集时，操作按照以下顺序进行： 1. Windows Kinesis 代理扫描安装目录中的程序集（`%PROGRAMFILES%\Amazon\AWSKinesisTap`），用于实现`IFactory`接口中定义`Amazon.KinesisTap.Core`程序集。此接口定义在`Amazon.KinesisTap.Core\Infrastructure\IFactory.cs`在 Kinesis 代理程序的 Windows 源代码中。 1. Windows 的 Kinesis 代理加载包含这些类的程序集并调用`RegisterFactory`方法。 1. 适用于 Windows 的 Kinesis 代理加载`appsettings.json`配置文件。对于配置文件中的每个源和接收器，检查 `SourceType` 和 `SinkType` 键/值对。如果有工厂的注册名称与 `SourceType` 和 `SinkType` 键/值对中的值相同，则调用这些工厂上的 `CreateInstance` 方法。`CreateInstance` 方法将配置和其他信息作为 `IPluginContext` 对象传递。`CreateInstance` 方法负责配置和初始化插件。 要使插件正确工作，必须有创建插件的已注册工厂类，并且插件类本身必须已定义。 Windows Kinesis 代理位于[https://github.com/awslabs/kinesis-agent-windows](https://github.com/awslabs/kinesis-agent-windows)。 ## 为 Windows 插件工厂实施 Kinesis 代理 按照以下步骤来实施 Windows 插件工厂的 Kinesis 代理。 **为 Windows 插件工厂创建一个 Kinesis 代理** 1. 创建定位到 .NET Framework 4.6 的 C\# 库项目。 1. 添加对 `Amazon.KinesisTap.Core` 程序集的引用。此程序集位于`%PROGRAMFILES%\Amazon\AWSKinesisTap`目录之后的 Kinesis 代理进行 Windows 安装。 1. 使用 `NuGet` 安装 `Microsoft.Extensions.Configuration.Abstractions` 程序包。 1. 使用 `NuGet` 安装 `System.Reactive` 程序包。 1. 使用 `NuGet` 安装 `Microsoft.Extensions.Logging` 程序包。 1. 创建为源实施 `IFactory` 或为接收器实施 `IFactory` 的工厂类。添加 `RegisterFactory` 和 `CreateInstance` 方法。 例如，以下代码创建 Windows 插件工厂的 Kinesis 代理，该工厂创建生成随机数据的源： ``` using System; using Amazon.KinesisTap.Core; using Microsoft.Extensions.Configuration; namespace MyCompany.MySources { public class RandomSourceFactory : IFactory { public void RegisterFactory(IFactoryCatalog catalog) { catalog.RegisterFactory("randomsource", this); } public ISource CreateInstance(string entry, IPlugInContext context) { IConfiguration config = context.Configuration; switch (entry.ToLower()) { case "randomsource": string rateString = config["Rate"]; string maxString = config["Max"]; TimeSpan rate; int max; if (string.IsNullOrWhiteSpace(rateString)) { rate = TimeSpan.FromSeconds(30); } else { if (!TimeSpan.TryParse(rateString, out rate)) { throw new Exception($"Rate {rateString} is invalid for RandomSource."); } } if (string.IsNullOrWhiteSpace(maxString)) { max = 1000; } else { if (!int.TryParse(maxString, out max)) { throw new Exception($"Max {maxString} is invalid for RandomSource."); } } return new RandomSource(rate, max, context); default: throw new ArgumentException($"Source {entry} is not recognized.", entry); } } } } ``` 当您最终希望增强工厂以创建不同类型的实例时，`switch` 语句在 `CreateInstance` 方法中使用。 要创建接收器工厂（该工厂创建不执行任何操作的接收器），请使用类似于下文的类： ``` using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using Amazon.KinesisTap.Core; using Microsoft.Extensions.Configuration; namespace MyCompany.MySinks { public class NullSinkFactory : IFactory { public void RegisterFactory(IFactoryCatalog catalog) { catalog.RegisterFactory("nullsink", this); } public IEventSink CreateInstance(string entry, IPlugInContext context) { IConfiguration config = context.Configuration; switch (entry.ToLower()) { case "nullsink": return new NullSink(context); default: throw new Exception("Unrecognized sink type {entry}."); } } } } ``` ## 实现 Windows 插件 Kinesis 代理程序 按照以下步骤来实施 Windows 插件源的 Kinesis 代理。 **创建 Windows 插件 Kinesis 代理程序** 1. 将实施 `IEventSource` 接口的类添加到以前为源创建的项目。 例如，使用以下代码来定义生成随机数据的源： ``` using System; using System.Reactive.Subjects; using System.Timers; using Amazon.KinesisTap.Core; using Microsoft.Extensions.Logging; namespace MyCompany.MySources { public class RandomSource : EventSource, IDisposable { private TimeSpan _rate; private int _max; private Timer _timer = null; private Random _random = new Random(); private ISubject> _recordSubject = new Subject>(); public RandomSource(TimeSpan rate, int max, IPlugInContext context) : base(context) { _rate = rate; _max = max; } public override void Start() { try { CleanupTimer(); _timer = new Timer(_rate.TotalMilliseconds); _timer.Elapsed += (Object source, ElapsedEventArgs args) => { var data = new RandomData() { RandomValue = _random.Next(_max) }; _recordSubject.OnNext(new Envelope(data)); }; _timer.AutoReset = true; _timer.Enabled = true; _logger?.LogInformation($"Random source id {this.Id} started with rate {_rate.TotalMilliseconds}."); } catch (Exception e) { _logger?.LogError($"Exception during start of RandomSource id {this.Id}: {e}"); } } public override void Stop() { try { CleanupTimer(); _logger?.LogInformation($"Random source id {this.Id} stopped."); } catch (Exception e) { _logger?.LogError($"Exception during stop of RandomSource id {this.Id}: {e}"); } } private void CleanupTimer() { if (_timer != null) { _timer.Enabled = false; _timer?.Dispose(); _timer = null; } } public override IDisposable Subscribe(IObserver> observer) { return this._recordSubject.Subscribe(observer); } public void Dispose() { CleanupTimer(); } } } ``` 在本示例中，`RandomSource` 类继承自 `EventSource` 类，因为它提供了 `Id` 属性。虽然本示例不支持书签，此基类也可用于实施该功能。信封提供了一种存储元数据和包装任意数据用于流式传输到接收器的方法。`RandomData` 类在下一步中定义，表示来自此源的输出对象的类型。 1. 添加类到以前定义的项目，该项目中包含从源流式传输的数据。 例如，随机数据的容器可以定义如下所示： ``` namespace MyCompany.MySources { public class RandomData { public int RandomValue { get; set; } } } ``` 1. 编译以前定义的项目。 1. 将程序集复制到 Windows 的 Kinesis 代理的安装目录。 1. 创建或更新`appsettings.json`配置文件，并将其放在 Windows 的 Kinesis 代理的安装目录中。 1. 停止然后重新启动 Windows 的 Kinesis 代理。 1. 检查当前 Kinesis 代理的 Windows 日志文件（通常位于`%PROGRAMDATA%\Amazon\AWSKinesisTap\logs`目录）以确保自定义源插件没有问题。 1. 确保数据到达目标 AWS 服务。 有关如何扩展`DirectorySource`功能来实施解析特定日志格式的功能，请参阅`Amazon.KinesisTap.Uls\UlsSourceFactory.cs`和`Amazon.KinesisTap.Uls\UlsLogParser.cs`在 Kinesis 代理程序的 Windows 源代码中。 有关如何创建提供书签功能的源的示例，请参阅`Amazon.KinesisTap.Windows\WindowsSourceFactory.cs`和`Amazon.KinesisTap.Windows\EventLogSource.cs`在 Kinesis 代理程序的 Windows 源代码中。 ## 实现 Windows 插件接收器的 Kinesis 代理 按照以下步骤来实施 Windows 插件接收器的 Kinesis 代理。 **为 Windows 插件接收器创建 Kinesis 代理程序** 1. 将类添加到之前定义的项目（该项目实施 `IEventSink` 接口）。 例如，以下代码实施接收器，该接收器除了对记录到达进行日志记录之外不执行任何操作，随后丢弃到达的记录。 ``` using Amazon.KinesisTap.Core; using Microsoft.Extensions.Logging; namespace MyCompany.MySinks { public class NullSink : EventSink { public NullSink(IPlugInContext context) : base(context) { } public override void OnNext(IEnvelope envelope) { _logger.LogInformation($"Null sink {Id} received {GetRecord(envelope)}."); } public override void Start() { _logger.LogInformation($"Null sink {Id} starting."); } public override void Stop() { _logger.LogInformation($"Null sink {Id} stopped."); } } } ``` 在此示例中，`NullSink` 接收器类继承自 `EventSink` 类，因为它提供将记录转换为不同序列化格式（例如 JSON 和 XML）的功能。 1. 编译以前定义的项目。 1. 将程序集复制到 Windows 的 Kinesis 代理的安装目录。 1. 创建或更新`appsettings.json`配置文件，并将其放在 Windows 的 Kinesis 代理的安装目录中。例如，要使用 `RandomSource` 和 `NullSink` 自定义插件，您可以使用以下 `appsettings.json` 配置文件： ``` { "Sources": [ { "Id": "MyRandomSource", "SourceType": "RandomSource", "Rate": "00:00:10", "Max": 50 } ], "Sinks": [ { "Id": "MyNullSink", "SinkType": "NullSink", "Format": "json" } ], "Pipes": [ { "Id": "MyRandomToNullPipe", "SourceRef": "MyRandomSource", "SinkRef": "MyNullSink" } ] } ``` 此配置创建发送 `RandomData` 的实例的源，其中 `RandomValue` 每隔 10 秒设置为 0 到 50 之间的随机数字。它创建一个接收器，将传入 `RandomData` 实例转换为 JSON、记录该 JSON 然后丢弃实例。请确保包含之前所定义项目中的示例工厂、`RandomSource` 源类以及 `NullSink` 接收器类以使用示例配置文件。 1. 停止然后重新启动 Windows 的 Kinesis 代理。 1. 检查当前 Kinesis 代理的 Windows 日志文件（通常位于`%PROGRAMDATA%\Amazon\AWSKinesisTap\logs`目录）以确保自定义接收器插件没有问题。 1. 确保数据到达目标 AWS 服务。由于示例 `NullSink` 不流式传输到 AWS 服务，您可以通过查找指示已收到记录的日志消息来验证接收器的操作正确。 举例来说，您可以看到类似于下文的日志文件： ``` 2018-10-18 12:36:36.3647 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.AWS.AWSEventSinkFactory. 2018-10-18 12:36:36.4018 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Windows.PerformanceCounterSinkFactory. 2018-10-18 12:36:36.4018 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory MyCompany.MySinks.NullSinkFactory. 2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Core.DirectorySourceFactory. 2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.ExchangeSource.ExchangeSourceFactory. 2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Uls.UlsSourceFactory. 2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Windows.WindowsSourceFactory. 2018-10-18 12:36:36.6926 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory MyCompany.MySources.RandomSourceFactory. 2018-10-18 12:36:36.9601 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.Core.Pipes.PipeFactory. 2018-10-18 12:36:37.4694 Amazon.KinesisTap.Hosting.LogManager INFO Registered factory Amazon.KinesisTap.AutoUpdate.AutoUpdateFactory. 2018-10-18 12:36:37.4807 Amazon.KinesisTap.Hosting.LogManager INFO Performance counter sink started. 2018-10-18 12:36:37.6250 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink starting. 2018-10-18 12:36:37.6250 Amazon.KinesisTap.Hosting.LogManager INFO Connected source MyRandomSource to sink MyNullSink 2018-10-18 12:36:37.6333 Amazon.KinesisTap.Hosting.LogManager INFO Random source id MyRandomSource started with rate 10000. 2018-10-18 12:36:47.8084 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":14}. 2018-10-18 12:36:57.6339 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":5}. 2018-10-18 12:37:07.6490 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":9}. 2018-10-18 12:37:17.6494 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":47}. 2018-10-18 12:37:27.6520 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":25}. 2018-10-18 12:37:37.6676 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":21}. 2018-10-18 12:37:47.6688 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":29}. 2018-10-18 12:37:57.6700 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":22}. 2018-10-18 12:38:07.6838 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":32}. 2018-10-18 12:38:17.6848 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":12}. 2018-10-18 12:38:27.6866 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":46}. 2018-10-18 12:38:37.6880 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":48}. 2018-10-18 12:38:47.6893 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":39}. 2018-10-18 12:38:57.6906 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":18}. 2018-10-18 12:39:07.6995 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":6}. 2018-10-18 12:39:17.7004 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":0}. 2018-10-18 12:39:27.7021 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":3}. 2018-10-18 12:39:37.7023 Amazon.KinesisTap.Hosting.LogManager INFO Null sink MyNullSink received {"RandomValue":19}. ``` 如果您创建访问 AWS 服务的接收器，则可能会发现基类非常有用。对于使用`AWSBufferedEventSink`基类，请参阅`Amazon.KinesisTap.AWS\CloudWatchLogsSink.cs`在适用于 Windows 的 Kinesis 代理的源代码中。