Introduzione ai plugin Kinesis Agent per Windows Implementazione dell'agente Kinesis per le fabbriche di plugin Windows Implementazione di Kinesis Agent per origini plugin Windows Implementazione di Kinesis Agent per i sink plugin di Windows

Creazione di Kinesis Agent per i plugin di Windows

Per la maggior parte delle situazioni, la creazione di un plugin Amazon Kinesis Agent per Microsoft Windows non è necessaria. Kinesis Agent per Windows è altamente configurabile e contiene origini e sink potenti, comeDirectorySourceeKinesisStream, che sono sufficienti per la maggior parte degli scenari. Per ulteriori informazioni sulle origini e i sink esistenti, consulta Configurazione di Amazon Kinesis Agent per Microsoft Windows.

Per gli scenari insoliti, potrebbe essere necessario estendere Kinesis Agent per Windows utilizzando un plug-in personalizzato. Di seguito sono riportati alcuni scenari di esempio:

La creazione di un pacchetto di una dichiarazione DirectorySource complessa utilizzando i parser di record Regex o Delimited in modo che sia facile da applicare in diversi tipi di file di configurazione.
Creazione di una nuova origine che non si basa su file o che supera la funzionalità di analisi fornita dal parser di record esistente.
Creazione di un sink per un servizio AWS che non è attualmente supportato.

Argomenti

Introduzione ai plugin Kinesis Agent per Windows
Implementazione dell'agente Kinesis per le fabbriche di plugin Windows
Implementazione di Kinesis Agent per origini plugin Windows
Implementazione di Kinesis Agent per i sink plugin di Windows

Introduzione ai plugin Kinesis Agent per Windows

Non c'è nulla di speciale sui plug-in personalizzati. Tutte le origini e i sink esistenti utilizzano gli stessi meccanismi che i plug-in personalizzati utilizzano per caricare quando viene avviato e creano un'istanza di plug-in rilevanti dopo la lettura delappsettings.jsonFile di configurazione.

Quando si avvia Kinesis Agent per Windows, si verifica la sequenza seguente:

Kinesis Agent per Windows analizza i gruppi nella directory di installazione (%PROGRAMFILES%\Amazon\AWSKinesisTap) per le classi che implementano ilIFactory<T>definita nellaAmazon.KinesisTap.CoreAssembly. Questa interfaccia è definita inAmazon.KinesisTap.Core\Infrastructure\IFactory.csnel codice sorgente Kinesis Agent per Windows.
Kinesis Agent per Windows carica gli assembly contenenti queste classi e richiama ilRegisterFactorysu queste classi.
Kinesis Agent per Windows carica ilappsettings.jsonFile di configurazione. Per ogni origine e sink nel file di configurazione, vengono esaminate le coppie chiave-valore SourceType e SinkType. Se non ci sono factory registrate con lo stesso nome dei valori delle coppie chiave-valore SourceType e SinkType, il metodo CreateInstance viene richiamato su tali factory. Il metodo CreateInstance viene passato alla configurazione e ad altre informazioni come un oggetto IPluginContext. Il metodo CreateInstance è responsabile della configurazione e inizializzazione del plug-in.

Affinché un plug-in funzioni correttamente, ci deve essere una factory registrata che crea il plug-in e deve essere definita la classe di plug-in.

Il codice sorgente di Kinesis Agent per Windows si trova inhttps://github.com/awslabs/kinesis-agent-windows: .

Implementazione dell'agente Kinesis per le fabbriche di plugin Windows

Segui questi passaggi per implementare una factory di plug-in di Kinesis per Windows.

Per creare una fabbrica di plugin Kinesis Agent per Windows

Creazione di un progetto di libreria C# indirizzato a .NET Framework 4.6.
Aggiungi un riferimento all'assembly Amazon.KinesisTap.Core. Questo assembly si trova nella%PROGRAMFILES%\Amazon\AWSKinesisTapdopo l'installazione di Kinesis Agent per Windows.
Utilizzare NuGet per installare il pacchetto Microsoft.Extensions.Configuration.Abstractions.
Utilizzare NuGet per installare il pacchetto System.Reactive.
Utilizzare NuGet per installare il pacchetto Microsoft.Extensions.Logging.

Creare una classe factory che implementa sia IFactory<IEventSource> per origini o IFactory<IEventSink> per sink. Aggiungere i metodi RegisterFactory e CreateInstance.

Ad esempio, il codice seguente crea una factory di plug-in di Kinesis Agent per Windows che crea un'origine che genera dati casuali:


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

La dichiarazione switch viene utilizzata nel metodo CreateInstance nel caso in cui si desidera migliorare la factory per creare diversi tipi di istanze.

Per creare una factory di sink che crea un sink che non fa nulla, utilizzare una classe simile alla seguente:


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

Implementazione di Kinesis Agent per origini plugin Windows

Segui questi passaggi per implementare un'origine di plug-in Kinesis per Windows.

Per creare un plugin Kinesis Agent per Windows

Aggiungi una classe che implementa l'interfaccia IEventSource<out T> al progetto creato in precedenza per l'origine.

Ad esempio, utilizzare il codice seguente per definire una sorgente che genera dati casuali:


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 questo esempio, la classe RandomSource eredita dalla classe EventSource<T> perché fornisce la proprietà Id. Anche se questo esempio non supporta i segnalibri, questa classe di base è utile anche per l'implementazione di tale funzionalità. Le envelope forniscono un modo per memorizzare i metadati e avvolgere i dati arbitrari per lo streaming ai sink. La classe RandomData viene definita nella fase successiva e rappresenta il tipo di oggetto di output da questa origine.

Aggiungi una classe al progetto definito in precedenza che contiene i dati oggetto di streaming dall'origine.

Ad esempio, un contenitore di dati casuali potrebbe essere definito come segue:
```
namespace MyCompany.MySources
{
    public class RandomData
    {
        public int RandomValue { get; set; }
    }
}               
```
Compila il progetto definito in precedenza.
Copia l'assembly nella directory di installazione per Kinesis Agent per Windows.
Crea o aggiorna unappsettings.jsonche utilizza la nuova origine e inserirlo nella directory di installazione per Kinesis Agent per Windows.
Arresto e avvio di Kinesis Agent per Windows.
Controllare il file di registro corrente di Kinesis Agent per Windows (in genere si trova nella%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) per garantire che non ci siano problemi con il plug-in sorgente personalizzato.
Verificare che i dati siano pervenuti al servizio AWS desiderato.

Per un esempio di come estendere ilDirectorySourcePer implementare l'analisi di un determinato formato di log, consultaAmazon.KinesisTap.Uls\UlsSourceFactory.cseAmazon.KinesisTap.Uls\UlsLogParser.csnel codice sorgente Kinesis Agent per Windows.

Per un esempio di come creare un'origine che fornisce la funzionalità di segnalibro, consultaAmazon.KinesisTap.Windows\WindowsSourceFactory.cseAmazon.KinesisTap.Windows\EventLogSource.csnel codice sorgente Kinesis Agent per Windows.

Implementazione di Kinesis Agent per i sink plugin di Windows

Segui questi passaggi per implementare il sink di plug-in Kinesis per Windows.

Per creare un sink di plugin Kinesis Agent per Windows

Aggiungi una classe al progetto definito in precedenza che implementa l'interfaccia IEventSink.

Ad esempio, il codice seguente implementa un sink che non fa altro che registrare l'arrivo dei record, in seguito rimossi.


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 questo esempio, la classe del sink NullSink eredita dalla classe EventSink perché offre la possibilità di trasformare i record in diversi formati di serializzazione, ad esempio JSON e XML.

Compila il progetto definito in precedenza.
Copia l'assembly nella directory di installazione per Kinesis Agent per Windows.
Crea o aggiorna unappsettings.jsonche utilizza il nuovo sink e inserirlo nella directory di installazione per Kinesis Agent per Windows. Ad esempio, per utilizzare i plug-in personalizzati RandomSource e NullSink, è possibile utilizzare il seguente file di configurazione 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"
	}
  ]
}               
```
Questa configurazione crea un'origine che invia un'istanza di RandomData con un set RandomValue a un numero casuale compreso tra 0 e 50 ogni 10 secondi. Crea un sink che trasforma le istanze RandomData in entrata per JSON, registra quel JSON, quindi elimina le istanze. Assicurati di includere entrambe le factory di esempio, la classe di origine RandomSource e la classe del sink NullSink nel progetto definito in precedenza per l'utilizzo di file di configurazione di esempio.
Arresto e avvio di Kinesis Agent per Windows.
Controllare il file di registro corrente di Kinesis Agent per Windows (in genere si trova nella%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) per garantire che non ci siano problemi con il plug-in sink personalizzato.

Verificare che i dati siano pervenuti al servizio AWS desiderato. Poiché l'esempio NullSink non viene trasmesso a un servizio AWS, è possibile verificare il corretto funzionamento del sink ricercando i messaggi di log che indicano che i record sono stati ricevuti.

Ad esempio, il file di log dovrebbe essere simile a quanto segue:


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

Se si crea un sink che accede ai servizi AWS, vi sono classi di base che possono risultare utili. Per un sink che utilizza ilAWSBufferedEventSinkclasse base, vedereAmazon.KinesisTap.AWS\CloudWatchLogsSink.csnel codice sorgente di Kinesis Agent per Windows.

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti

Risoluzione dei problemi

Cronologia dei documenti