Syslog-NG Python Loki Plugin

loki.py

import grpc

import logproto_pb2
import logproto_pb2_grpc

import google.protobuf.timestamp_pb2
import time

print("=== loaded loki plugin ===")

class Loki(object):

    def open(self):
        """Open a connection to the target service
        Should return False if opening fails"""

        return True

    def close(self):
        """Close the connection to the target service"""
        pass

    def is_opened(self):
        """Check if the connection to the target is able to receive messages"""
        return True

    def init(self, options):
        """This method is called at initialization time

        Should return false if initialization fails"""

        self.channel = grpc.insecure_channel('localhost:9095')
        self.stub = logproto_pb2_grpc.PusherStub(self.channel)

        return True

    def deinit(self):
        """This method is called at deinitialization time"""
        pass

    def send(self, msg):
        host = msg["HOST"]
        program = msg["PROGRAM"]
        labels = '{job="syslog-ng",host="'+host+'",program="'+program+'"}'

        ts = google.protobuf.timestamp_pb2.Timestamp();
        ts.GetCurrentTime();

        entry = logproto_pb2.Entry(timestamp=ts,line=msg["MESSAGE"])

        streams = [logproto_pb2.Stream(labels=labels,entries=[entry])]
        response = self.stub.Push(logproto_pb2.PushRequest(streams=streams))

        """Send a message to the target service

        It should return True to indicate success. False will suspend the
        destination for a period specified by the time-reopen() option."""
        return True

loki_destination.conf

destination d_python_to_loki {
    python(
        class("loki.Loki")
        value-pairs(scope(nv_pairs core syslog all_macros selected_macros everything))
    );
};
log {
    source(s_net);
    destination(d_python_to_loki);
};