yu000hong / flumehandler Goto Github PK

[2020-08-13 10:46:43] root ERROR Error when sending events to flume, host is xx.xx.xx.xx
Traceback (most recent call last):
File "flume_agent.py", line 115, in _send
client.appendBatch(events)
File "thrift_protocol.py", line 83, in appendBatch
return self.recv_appendBatch()
File "thrift_protocol.py", line 95, in recv_appendBatch
(fname, mtype, rseqid) = iprot.readMessageBegin()
File "/usr/lib64/python3.6/site-packages/thrift/protocol/TCompactProtocol.py", line 334, in readMessageBegin
proto_id = self.__readUByte()
File "/usr/lib64/python3.6/site-packages/thrift/protocol/TCompactProtocol.py", line 313, in __readUByte
result, = unpack('!B', self.trans.readAll(1))
File "/usr/lib64/python3.6/site-packages/thrift/transport/TTransport.py", line 62, in readAll
chunk = self.read(sz - have)
File "/usr/lib64/python3.6/site-packages/thrift/transport/TTransport.py", line 283, in read
self.readFrame()
File "/usr/lib64/python3.6/site-packages/thrift/transport/TTransport.py", line 287, in readFrame
buff = self.__trans.readAll(4)
File "/usr/lib64/python3.6/site-packages/thrift/transport/TTransport.py", line 62, in readAll
chunk = self.read(sz - have)
File "/usr/lib64/python3.6/site-packages/thrift/transport/TSocket.py", line 126, in read
buff = self.handle.recv(sz)
File "/usr/lib64/python3.6/site-packages/gevent/_socket3.py", line 382, in recv
self._wait(self._read_event)
File "src/gevent/_hub_primitives.py", line 284, in gevent.__hub_primitives.wait_on_socket
File "src/gevent/_hub_primitives.py", line 289, in gevent.__hub_primitives.wait_on_socket
File "src/gevent/_hub_primitives.py", line 264, in gevent.__hub_primitives._primitive_wait
gevent.exceptions.ConcurrentObjectUseError: This socket is already used by another greenlet: <bound method Waiter.switch of <gevent.__waiter.Waiter object at 0x7fae4d203048>>

record.args is not always of dict type, it may be a tuple:

logger.info("%s, %s", ("hello", "world"))

def convert(self, record):
        headers = self.evaluate(self.headers)
        record.args.update(self.evaluate(self.envs))
        body = bytes(self.format(record), 'utf8')
        return ThriftFlumeEvent(headers=headers, body=body)

The args attribute of LogRecord has been changed!

    def evaluate(self, d, record):
        result = dict()
        for k, v in d.items():
            if callable(v):
                args_len = len(getargspec(v).args)
                if args_len <= 0:
                    result[k] = v()
                elif args_len == 1:
                    result[k] = v(record.args.get(k))
                else:
                    result[k] = v(record.args.get(k), record.args)
            else:
                result[k] = v
        return result

If the env value is a function, it can accept zero/one/two parameters.

If the value function accept one parameter, it should be the value which has been in record.args.

If the value function accept two parameters, the first should be the value which has been in record.args, and the second is just record.args itself.

So, we can compute value from its old value or all values in record.args.