Erstellen von Kinesis Agent für Windows-Plugins

Erstellen Sie ein C#-Bibliotheksprojekt für .NET Framework 4.6.

Fügen Sie einen Verweis auf die Komponente Amazon.KinesisTap.Core hinzu. Diese Baugruppe befindet sich im%PROGRAMFILES%\Amazon\AWSKinesisTap-Verzeichnis nach der Installation von Kinesis Agent für Windows.

Installieren Sie mithilfe von NuGet das Paket Microsoft.Extensions.Configuration.Abstractions.

Installieren Sie mithilfe von NuGet das Paket System.Reactive.

Installieren Sie mithilfe von NuGet das Paket Microsoft.Extensions.Logging.

Erstellen Sie eine Factory-Klasse, mit der entweder IFactory<IEventSource> für Quellen oder IFactory<IEventSink> für Senken implementiert wird. Fügen Sie die RegisterFactory- oder CreateInstance-Methoden hinzu.

Beispielsweise erstellt der folgende Code eine -Plug-In-Factory von Kinesis Agent für Windows, mit der eine Quelle erstellt wird, mit der zufällige Daten generiert werden:

using System;
using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Configuration;

namespace MyCompany.MySources
{
    public class RandomSourceFactory : IFactory<ISource>
    {
        public void RegisterFactory(IFactoryCatalog<ISource> 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);
            }
        }
    }
}               

Die switch-Anweisung wird in der Methode CreateInstance verwendet, falls Sie die Factory irgendwann zum Erstellen verschiedener Arten von Instances erweitern möchten.

Um eine Senken-Factory zu erstellen, mit der eine aktionslose Senke erstellt wird, verwenden Sie eine Klasse ähnlich der folgenden:

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<IEventSink>
    {
        public void RegisterFactory(IFactoryCatalog<IEventSink> 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}.");
            }
        }
    }
}             

Fügen Sie eine Klasse hinzu, mit der die IEventSource<out T>-Schnittstelle zu dem zuvor für die Quelle erstellten Projekt hinzufügt wird.

Verwenden Sie z. B. den folgenden Code, um eine Quelle zu definieren, mit der zufällige Daten generiert werden:

using System;
using System.Reactive.Subjects;
using System.Timers;
using Amazon.KinesisTap.Core;
using Microsoft.Extensions.Logging;

namespace MyCompany.MySources
{
    public class RandomSource : EventSource<RandomData>, IDisposable
    {
        private TimeSpan _rate;
        private int _max;
        private Timer _timer = null;
        private Random _random = new Random();
        private ISubject<IEnvelope<RandomData>> _recordSubject = new Subject<IEnvelope<RandomData>>();
        


        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<RandomData>(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<IEnvelope<RandomData>> observer)
        {
            return this._recordSubject.Subscribe(observer);
        }

        public void Dispose()
        {
            CleanupTimer();
        }
    }
}               

In diesem Beispiel erbt die Klasse RandomSource von der Klasse EventSource<T>, da sie die Eigenschaft Id bereitstellt. Obwohl dieses Beispiel nicht das Setzen von Lesezeichen unterstützt, ist diese Basisklasse auch für die Implementierung dieser Funktionalität nützlich. Envelopes bieten eine Möglichkeit zum Speichern von Metadaten und Verpacken beliebiger Daten zum Streamen an Senken. Die Klasse RandomData wird im nächsten Schritt definiert und stellt den Typ des Ausgabeobjekts aus dieser Quelle dar.

Fügen Sie eine Klasse zu dem zuvor definierten Projekt hinzu, das die Daten enthält, die aus der Quelle gestreamt werden.

Beispiel: Ein Container für zufällige Daten könnte wie folgt definiert werden:

namespace MyCompany.MySources
{
    public class RandomData
    {
        public int RandomValue { get; set; }
    }
}               

Kompilieren Sie das zuvor definierte Projekt.

Kopieren Sie die Komponente in das Installationsverzeichnis für Kinesis Agent für Windows.

Erstellen oder aktualisierenappsettings.jsonFühren Sie die -Konfigurationsdatei durch, und platzieren Sie sie im Installationsverzeichnis für Kinesis Agent für Windows.

Kinesis Agent für Windows beenden und starten.

Überprüfen Sie die aktuelle Kinesis Agent für Windows-Protokolldatei (normalerweise im%PROGRAMDATA%\Amazon\AWSKinesisTap\logsFühren Sie das -Verzeichnis durch, um sicherzustellen, dass keine Probleme mit dem benutzerdefinierten Quell-Plug-In vorliegen.

Stellen Sie sicher, dass die Daten bei dem gewünschten AWS-Service eintreffen.

Fügen Sie zu dem zuvor definierten Projekt eine Klasse hinzu, mit der die IEventSink-Schnittstelle implementiert wird.

Beispiel: Der folgende Code implementiert eine Senke, die nichts anderes macht, als den Eingang von Datensätzen zu protokollieren, die dann verworfen werden.

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.");
        }
    }
}               

In diesem Beispiel erbt die NullSink-Senken-Klasse von der EventSink-Klasse, da sie die Möglichkeit bietet, Datensätze in verschiedene Serialisierungsformate, wie z. B. JSON und XML, zu transformieren.

Kompilieren Sie das zuvor definierte Projekt.

Kopieren Sie die Komponente in das Installationsverzeichnis für Kinesis Agent für Windows.

Erstellen oder aktualisierenappsettings.jsonFühren Sie die -Konfigurationsdatei durch, die von der neuen Senke Gebrauch macht, und platzieren Sie sie im Installationsverzeichnis für Kinesis Agent für Windows. Angenommen, Sie möchten die benutzerdefinierten Plug-Ins RandomSource und NullSink nutzen. In diesem Fall könnten Sie die folgende appsettings.json-Konfigurationsdatei verwenden:

{
  "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"
	}
  ]
}               

Diese Konfiguration erstellt eine Quelle, die eine Instance von RandomData sendet, bei der RandomValue alle 10 Sekunden auf eine Zufallszahl zwischen 0 und 50 eingestellt wird. Die hiermit erstellte Senke transformiert die eingehenden RandomData-Instances in das JSON-Format, protokolliert das JSON-Format und verwirft die Instances. Achten Sie darauf, beide Factories, die Quell-Klasse RandomSource und die Senken-Klasse NullSink, im zuvor definierten Projekt einzuschließen, um die Beispiel-Konfigurationsdatei verwenden zu können.

Kinesis Agent für Windows beenden und starten.

Überprüfen Sie die aktuelle Kinesis Agent für Windows-Protokolldatei (normalerweise im%PROGRAMDATA%\Amazon\AWSKinesisTap\logsFühren Sie die folgenden Schritt durch, um sicherzustellen, dass es keine Probleme mit dem benutzerdefinierten Senken-Plug-In gibt.

Stellen Sie sicher, dass die Daten bei dem gewünschten AWS-Service eintreffen. Da das Beispiel NullSink nicht an einen AWS-Service gestreamt wird, können Sie die richtige Operation der Senke anhand von Protokollmeldungen überprüfen, aus denen hervorgeht, dass die Datensätze empfangen wurden.

Beispielsweise kann die angezeigte Protokolldatei wie folgt aussehen:

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}.