Introducción a los complementos de Kinesis Agent para Windows Implementación del agente Kinesis para fábricas de complementos de Windows Implementación del agente Kinesis para fuentes de complementos de Windows Implementación del agente Kinesis para sumideros de complementos de Windows

Creación de Kinesis Agent para complementos de Windows

En la mayoría de las situaciones, no es necesario crear un complemento de Amazon Kinesis Agent para Microsoft Windows. Kinesis Agent para Windows tiene una gran capacidad de configuración y dispone de orígenes y receptores muy eficaces, comoDirectorySourceyKinesisStream, que son suficientes para la mayoría de los escenarios. Para obtener más información sobre los orígenes y receptores existentes, consulte Configuración de Amazon Kinesis Agent para Microsoft Windows.

En escenarios poco comunes, puede ser necesario ampliar el agente de Kinesis para Windows con un complemento personalizado. Algunos de estos escenarios podrían ser los siguientes:

Se está empaquetando una declaración DirectorySource compleja con analizadores de recursos Regex o Delimited para que resulte fácil aplicarla en diferentes tipos de archivos de configuración.
Se está creando un nuevo origen que no está basado en un archivo o que sobrepasa las funcionalidades de análisis que proporcionan los analizadores de registros existentes.
Se está creando un receptor para un servicio de AWS que actualmente no es compatible.

Temas

Introducción a los complementos de Kinesis Agent para Windows
Implementación del agente Kinesis para fábricas de complementos de Windows
Implementación del agente Kinesis para fuentes de complementos de Windows
Implementación del agente Kinesis para sumideros de complementos de Windows

Introducción a los complementos de Kinesis Agent para Windows

No hay nada especial en los complementos personalizados. Todos los orígenes y receptores existentes utilizan los mismos mecanismos que usan los complementos personalizados para cargarse cuando se inicia el agente Kinesis para Windows y crean instancias de los complementos apropiados después de leer elappsettings.jsonArchivo de configuración.

Cuando se inicia el agente Kinesis para Windows, se produce la siguiente secuencia:

Kinesis Agent para Windows analiza los ensamblados en el directorio de instalación (%PROGRAMFILES%\Amazon\AWSKinesisTap) para las clases que implementan elIFactory<T>definida en la interfazAmazon.KinesisTap.CoreAssembly EnAmazon.KinesisTap.Core\Infrastructure\IFactory.csen el código fuente de Kinesis Agent para Windows.
Kinesis Agent para Windows carga los ensamblados que contienen estas clases e invoca el métodoRegisterFactoryen estas clases.
Kinesis Agent para Windows carga elappsettings.jsonArchivo de configuración. En cada origen y receptor del archivo de configuración, se examinan los pares clave-valor SourceType y SinkType. Si hay factorías registradas con el mismo nombre que los valores de los pares clave-valor SourceType y SinkType, se invoca el método CreateInstance en dichas factorías. El método CreateInstance recibe la configuración y otra información como un objeto IPluginContext. El método CreateInstance es responsable de configurar e inicializar el complemento.

Para que un complemento funcione correctamente, debe haber una clase de factoría registrada que cree el complemento y debe definirse la propia clase del complemento.

El código fuente de Kinesis Agent para Windows se encuentra enhttps://github.com/awslabs/kinesis-agent-windows.

Implementación del agente Kinesis para fábricas de complementos de Windows

Siga estos pasos para implementar una factoría de complementos de Kinesis Agent para Windows.

Para crear una fábrica de complementos de Kinesis Agent para Windows

Cree un proyecto de biblioteca de C# que tenga como destino .NET Framework 4.6.
Añada una referencia al ensamblado Amazon.KinesisTap.Core. Este ensamblado se encuentra en el%PROGRAMFILES%\Amazon\AWSKinesisTapdespués de la instalación de Kinesis Agent para Windows.
Utilice NuGet para instalar el paquete Microsoft.Extensions.Configuration.Abstractions.
Utilice NuGet para instalar el paquete System.Reactive.
Utilice NuGet para instalar el paquete Microsoft.Extensions.Logging.

Cree una clase de factoría que implemente IFactory<IEventSource> para los orígenes o IFactory<IEventSink> para los receptores. Añada los métodos CreateInstance y RegisterFactory.

Por ejemplo, el siguiente código crea una factoría de complementos de Kinesis Agent para Windows que crea un origen que genera datos aleatorios:


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 instrucción switch se utilizará en el método CreateInstance en caso de que finalmente desee mejorar la factoría para crear diferentes tipos de instancias.

Para crear un factoría de receptores que cree un receptor que no haga nada, utilice una clase similar a la siguiente:


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

Implementación del agente Kinesis para fuentes de complementos de Windows

Siga estos pasos para implementar un origen de complementos de Kinesis Agent para Windows.

Para crear un origen de complemento de Kinesis Agent para Windows

Añada una clase que implemente la interfaz IEventSource<out T> al proyecto que creó con anterioridad para el origen.

Por ejemplo, utilice el siguiente código para definir un origen que genere datos aleatorios:


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

En este ejemplo, la clase RandomSource hereda de la clase EventSource<T>, ya que proporciona la propiedad Id. Aunque este ejemplo no admite marcadores, esta clase base también resulta útil para implementar esta funcionalidad. Los sobres constituyen un mecanismo para almacenar metadatos y envolver datos arbitrarios para transmitirlos a los receptores. La clase RandomData se define en el siguiente paso y representa el tipo de objeto de salida de este origen.

Añada al proyecto que definió anteriormente una clase que contenga los datos transmitidos desde el origen.

Por ejemplo, un contenedor de datos aleatorios podría definirse del modo siguiente:
```
namespace MyCompany.MySources
{
    public class RandomData
    {
        public int RandomValue { get; set; }
    }
}               
```
Compile el proyecto definido anteriormente.
Copie el ensamblado en el directorio de instalación de Kinesis Agent para Windows.
Cree o actualice unappsettings.jsonque utilice el nuevo origen y sitúelo en el directorio de instalación de Kinesis Agent para Windows.
Detenga e inicie Kinesis Agent para Windows.
Compruebe el archivo de registro de Kinesis Agent para Windows actual (normalmente se encuentra en la carpeta%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) para comprobar que no hay ningún problema con el complemento de origen personalizado.
Asegúrese de que llegan datos al servicio de AWS deseado.

Para ver un ejemplo de cómo ampliar elDirectorySourcePara implementar el análisis de un determinado formato de registro, consulteAmazon.KinesisTap.Uls\UlsSourceFactory.csyAmazon.KinesisTap.Uls\UlsLogParser.csen el código fuente de Kinesis Agent para Windows.

Para obtener un ejemplo acerca de cómo crear un origen que proporcione la funcionalidad de marcadores, consulteAmazon.KinesisTap.Windows\WindowsSourceFactory.csyAmazon.KinesisTap.Windows\EventLogSource.csen el código fuente de Kinesis Agent para Windows.

Implementación del agente Kinesis para sumideros de complementos de Windows

Siga estos pasos para implementar un receptor de complementos de Kinesis Agent para Windows.

Para crear un receptor del complemento de Kinesis Agent para Windows

Añada una clase que implemente la interfaz IEventSink al proyecto que definió anteriormente.

Por ejemplo, el código siguiente implementa un receptor que no hace nada más que registrar la llegada de entradas de registro y luego las descarta.


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

En este ejemplo, la clase del receptor NullSink hereda de la clase EventSink porque le brinda la posibilidad de transformar registros en diferentes formatos de serialización, como JSON y XML.

Compile el proyecto definido anteriormente.
Copie el ensamblado en el directorio de instalación de Kinesis Agent para Windows.
Cree o actualice unappsettings.jsonque utilice el nuevo receptor y sitúelo en el directorio de instalación de Kinesis Agent para Windows. Por ejemplo, para utilizar los complementos personalizados RandomSource y NullSink, puede utilizar el siguiente archivo de configuración 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"
	}
  ]
}               
```
Esta configuración crea un origen que envía una instancia de RandomData con un RandomValue establecido en un número aleatorio entre 0 y 50 cada 10 segundos. Crea un receptor que transforma las instancias de RandomData entrantes en JSON, registra este código JSON y después descarta las instancias. No olvide incluir las dos factorías del ejemplo, la clase del origen RandomSource la clase del receptor NullSink en el proyecto que definió anteriormente para poder utilizar el archivo de configuración del ejemplo.
Detenga e inicie Kinesis Agent para Windows.
Compruebe el archivo de registro de Kinesis Agent para Windows actual (normalmente se encuentra en la carpeta%PROGRAMDATA%\Amazon\AWSKinesisTap\logs) para comprobar que no hay ningún problema con el complemento del receptor personalizado.

Asegúrese de que llegan datos al servicio de AWS deseado. Como la clase NullSink de ejemplo no transmite datos a los servicios de AWS, puede comprobar que el receptor funciona correctamente buscando mensajes de registro que indiquen que se han recibido los registros.

Por ejemplo, podría ver un archivo de registro similar al siguiente:


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

Si crea un receptor con acceso a los servicios de AWS, hay clases base que podrían resultarle útiles. Para un fregadero que utiliza elAWSBufferedEventSinkclase base, consulteAmazon.KinesisTap.AWS\CloudWatchLogsSink.csen el código fuente de Kinesis Agent para Windows.

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Solución de problemas

Historial de revisión