kapacitor UDF getting stuck on batch tasks

Created on 12 Jun 2018  路  13Comments  路  Source: influxdata/kapacitor

Hi, we create several UDFs on several kapacitor servers and we encountered some issue repeating for different UDFs on different kapacitors.

For example we created some UDF with kapacitor task (they both described in the topic we created in influxdata community: https://community.influxdata.com/t/unexpected-eof-error-when-running-a-simple-udf-in-kapacitor/5456)

when the UDF sends the generated points back to the kapacitor ( self._agent.write_response(point_response) )
it seems it gets stuck.
After some debugging I see that the UDF is stuck in the self._out.write(data) in write_response.

It also seems that for some reason at some stage the kapacitor close the PIPE and stop reading the incoming points and that's why the self._out.write(data) is blocked
We still did not figured why it happends but when I open the PIPE again for reading (using cat /proc/kapacitor_pid/fd/
I see the remaining points are printed and the UDF is again unblocked.

The question is why is it getting blocked and why the points are not read anymore..

picture nirshar
👍3

Most helpful comment

Hi, I managed to solve this issue.
We tried to create our UDF in python3 based on changes in agent.py that we found in the community that make the UDF runnable in python3. But it was running with functionality issues.

What was actually happen is that we could see in the log that the UDF started to run and receive points but when it tried to send points out -
it got blocked at some point and never finish the task. like kind of deadlock.
The process was still running, there were no Exceptions but we started to receive "unexpected EOF" messages from the snpashot task and it never finish the processing.

After some tracing and debuging from the python agent lib, and from the kapacitor code,
I found that the issue was the encodeUvarint(writer, value) method.
This method use chr() build-in function to create 1 char string representing the byte value and send it to stdout. It works for python2.
In python3 it uses "encode" [chr(varintMoreMask|bits).encode()] to make it send byte instead of 1 char string to stdout.buffer.
The issue here is that encode use UTF-8 as default, which is variable width character encoding. so for every value > 127 it encodes the value to at least 2-bytes.
When the kapacitor tries to read and decode it, it has an extra bytes to read and they are decoded to completely different number.

For example, encodeUvarint(sys.stdout.buffer, 190) should write to stdout [0xBE, 0x01]. But instead it writes [0xC3, 0xBE, 0x01].
When the kapacitor tries to decode it, its decoded to 24386.
Now the kapacitor tries to read 24386 bytes, but it has only 190. Then it has "Unexpected EOF" and it closes the PIPE and stops reading from the stream.
The UDF agent doesn't know the stream is not been read anymore and continues to write data to stdout.buffer until it reach the max buffer size and then gets blocked.
At this stage the UDF is blocked in the write_response on the self._out.write(data) command and it stays on this state since the stdout buffer is full and no one reads it.

To solve it, I changed the encodeUvarint to use bytes() function instead chr().encode().
So instead:
writer.write(chr(varintMoreMask|bits).encode())
I use:
writer.write(bytes([varintMoreMask|bits])) 

def encodeUvarint(writer, value):
    bits = value & varintMask
    value >>= shiftSize
    while value:
        writer.write(bytes([varintMoreMask|bits]))
        bits = value & varintMask
        value >>= shiftSize
    return writer.write(bytes([bits]))

After this change the UDF was able to execute normally.

I currently have the same issue (same symptoms) on UDF with python2, I will update what was the issue there and how it solved once I have time to access there and solve it.

picture nirshar on 17 Jun 2018
🎉6

All 13 comments

Hi, I managed to solve this issue.
We tried to create our UDF in python3 based on changes in agent.py that we found in the community that make the UDF runnable in python3. But it was running with functionality issues.

What was actually happen is that we could see in the log that the UDF started to run and receive points but when it tried to send points out -
it got blocked at some point and never finish the task. like kind of deadlock.
The process was still running, there were no Exceptions but we started to receive "unexpected EOF" messages from the snpashot task and it never finish the processing.

After some tracing and debuging from the python agent lib, and from the kapacitor code,
I found that the issue was the encodeUvarint(writer, value) method.
This method use chr() build-in function to create 1 char string representing the byte value and send it to stdout. It works for python2.
In python3 it uses "encode" [chr(varintMoreMask|bits).encode()] to make it send byte instead of 1 char string to stdout.buffer.
The issue here is that encode use UTF-8 as default, which is variable width character encoding. so for every value > 127 it encodes the value to at least 2-bytes.
When the kapacitor tries to read and decode it, it has an extra bytes to read and they are decoded to completely different number.

For example, encodeUvarint(sys.stdout.buffer, 190) should write to stdout [0xBE, 0x01]. But instead it writes [0xC3, 0xBE, 0x01].
When the kapacitor tries to decode it, its decoded to 24386.
Now the kapacitor tries to read 24386 bytes, but it has only 190. Then it has "Unexpected EOF" and it closes the PIPE and stops reading from the stream.
The UDF agent doesn't know the stream is not been read anymore and continues to write data to stdout.buffer until it reach the max buffer size and then gets blocked.
At this stage the UDF is blocked in the write_response on the self._out.write(data) command and it stays on this state since the stdout buffer is full and no one reads it.

To solve it, I changed the encodeUvarint to use bytes() function instead chr().encode().
So instead:
writer.write(chr(varintMoreMask|bits).encode())
I use:
writer.write(bytes([varintMoreMask|bits])) 

def encodeUvarint(writer, value):
    bits = value & varintMask
    value >>= shiftSize
    while value:
        writer.write(bytes([varintMoreMask|bits]))
        bits = value & varintMask
        value >>= shiftSize
    return writer.write(bytes([bits]))

After this change the UDF was able to execute normally.

I currently have the same issue (same symptoms) on UDF with python2, I will update what was the issue there and how it solved once I have time to access there and solve it.

nirshar picture nirshar on 17 Jun 2018
🎉6

Good Job !

ninjadude333 picture ninjadude333 on 19 Jun 2018

Hello

we have the same issue there. Running python 2.7.
Did you finally solve the issue on 2.7 ?
thanks in advance
regards

garbamoussa picture garbamoussa on 30 Oct 2018

nope, we worked on python 3 with solving the issue as @nirshar describes above.

ninjadude333 picture ninjadude333 on 1 Nov 2018

2170 has the similar problem. use .encode() in the above mentioned function and it works

shantanoo-desai picture shantanoo-desai on 14 Mar 2019

Python37 on windows 10 i was not able to start the kapacitor server.
Changing encodeUvarint to use bytes() solved the issue and i was able to start the UDF.
Thank you guys!

ashishkaransingh picture ashishkaransingh on 2 Oct 2019
👍1

Now i am getting the following logs in kapacitor.log

kapacitor.conf

[udf]
[udf.functions]
    [udf.functions.pyavg]
        # Run python
        prog = "C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\python.exe"
        # Pass args to python
        # -u for unbuffered STDIN and STDOUT
        # and the path to the script
        args = ["-u", "C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\examples\\moving_avg\\moving_avg.py"]
        # If the python process is unresponsive for 10s kill it
        timeout = "10s"
        # Define env vars for the process, in this case the PYTHONPATH
        [udf.functions.pyavg.env]
            PYTHONPATH = "C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py"

ts=2019-10-02T20:47:33.791+05:30 lvl=debug msg="opening service" source=srv service=*udf.Service
ts=2019-10-02T20:47:34.003+05:30 lvl=info msg="UDF log" service=udf text="2019-10-02 20:47:34,003 INFO:root: Starting Agent"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="Traceback (most recent call last):"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="**UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1069, in MergeFromString"**
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    **if self._InternalParse(serialized, 0, length) != length:"**
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1091, in InternalParse"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    (tag_bytes, new_pos) = local_ReadTag(buffer, pos)"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\decoder.py\", line 181, in ReadTag"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    while six.indexbytes(buffer, pos) & 0x80:"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="TypeError: unsupported operand type(s) for &: 'str' and 'int'"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="During handling of the above exception, another exception occurred:"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="Traceback (most recent call last):"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py\\kapacitor\\udf\\agent.py\", line 99, in _read_loop"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    request.ParseFromString(data)"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\message.py\", line 185, in ParseFromString"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    self.MergeFromString(serialized)"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1075, in MergeFromString"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="    raise message_mod.DecodeError('Truncated message.')"
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="google.protobuf.message.DecodeError: Truncated message."
ts=2019-10-02T20:47:34.005+05:30 lvl=info msg="UDF log" service=udf text="2019-10-02 20:47:34,005 ERROR:root: error processing request of type unknown: Truncated message."
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="Exception in thread Thread-1:"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="Traceback (most recent call last):"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1069, in MergeFromString"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    if self._InternalParse(serialized, 0, length) != length:"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1091, in InternalParse"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    (tag_bytes, new_pos) = local_ReadTag(buffer, pos)"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\decoder.py\", line 181, in ReadTag"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    while six.indexbytes(buffer, pos) & 0x80:"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="TypeError: unsupported operand type(s) for &: 'str' and 'int'"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="During handling of the above exception, another exception occurred:"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="Traceback (most recent call last):"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py\\kapacitor\\udf\\agent.py\", line 99, in _read_loop"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    request.ParseFromString(data)"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\message.py\", line 185, in ParseFromString"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    self.MergeFromString(serialized)"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\site-packages\\google\\protobuf\\internal\\python_message.py\", line 1075, in MergeFromString"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    raise message_mod.DecodeError('Truncated message.')"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="google.protobuf.message.DecodeError: Truncated message."
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="During handling of the above exception, another exception occurred:"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="Traceback (most recent call last):"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\threading.py\", line 926, in _bootstrap_inner"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    self.run()"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\Users\\testuser\\AppData\\Local\\Programs\\Python\\Python37\\lib\\threading.py\", line 870, in run"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    self._target(*self._args, **self._kwargs)"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py\\kapacitor\\udf\\agent.py\", line 135, in _read_loop"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    self.write_response(response)"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py\\kapacitor\\udf\\agent.py\", line 82, in write_response"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    encodeUvarint(self._out, len(data))"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="  File \"C:\\temp\\kapacitor_udf\\kapacitor\\udf\\agent\\py\\kapacitor\\udf\\agent.py\", line 158, in encodeUvarint"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="    return writer.write(bytes(bits))"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="TypeError: write() argument must be str, not bytes"
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text=
ts=2019-10-02T20:47:34.007+05:30 lvl=info msg="UDF log" service=udf text="2019-10-02 20:47:34,007 INFO:root: Agent finished"
ashishkaransingh picture ashishkaransingh on 2 Oct 2019

@shantanoo-desai I looked again carefully and the log is similar to yours in #2170
Where can i access the patch you mentioned of:

1750 patch solves this issue. However keeping it open currently for further tests

ashishkaransingh picture ashishkaransingh on 2 Oct 2019

@ashishkaransingh can You simply browse through the blog here where agent.py is mentioned. You can either patch the code or just change it by editing the file respectively.

shantanoo-desai picture shantanoo-desai on 2 Oct 2019
👍1

Awesome thanks a lot @shantanoo-desai !

ashishkaransingh picture ashishkaransingh on 3 Oct 2019

I am able to start kapacitor and see Listening signals.
But when i enabled my task which is using moving_avg.py from ("../kapacitor\udf\agent\examples\moving_avg") i have started getting the following errors

image

ts=2019-10-03T21:57:00.323+05:30 lvl=info msg=point service=kapacitor task_master=main task=udf_M_Average node=log3 prefix= name=Processor db=influxdb_test rp=autogen group=  tag_host=A700459-W10 tag_instance=_Total tag_objectname=Processor field_Percent_Idle_Time=80.53103637695312 time=2019-10-03T16:27:00Z
ts=2019-10-03T21:57:00.323+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=pyavg4 child=log5 collected=0 emitted=0
ts=2019-10-03T21:57:00.323+05:30 lvl=error msg="node failed" service=kapacitor task_master=main task=udf_M_Average node=pyavg4 err="keepalive timedout, last keepalive received was: 2019-10-03 21:55:57.0707919 +0530 IST"
ts=2019-10-03T21:57:00.323+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=log5 child=http_out6 collected=0 emitted=0
ts=2019-10-03T21:57:00.323+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=http_out6 child=influxdb_out7 collected=0 emitted=0
ts=2019-10-03T21:57:00.323+05:30 lvl=debug msg="task finished" service=task_store task=udf_M_Average
ts=2019-10-03T21:57:00.323+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=stream child=stream0 collected=219 emitted=219
ts=2019-10-03T21:57:00.324+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=stream0 child=from1 collected=219 emitted=219
ts=2019-10-03T21:57:00.324+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=from1 child=http_out2 collected=25 emitted=25
ts=2019-10-03T21:57:00.324+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=http_out2 child=log3 collected=25 emitted=25
ts=2019-10-03T21:57:00.324+05:30 lvl=debug msg="closing edge" service=kapacitor task_master=main task=udf_M_Average parent=log3 child=pyavg4 collected=25 emitted=25
ts=2019-10-03T21:57:00.324+05:30 lvl=error msg="failed to stop task with out error" service=kapacitor task_master=main task=udf_M_Average err="pyavg4: keepalive timedout, last keepalive received was: 2019-10-03 21:55:57.0707919 +0530 IST"
ts=2019-10-03T21:57:00.324+05:30 lvl=error msg="task finished with error" service=task_store err="pyavg4: keepalive timedout, last keepalive received was: 2019-10-03 21:55:57.0707919 +0530 IST" task=udf_M_Average

My Tickscript

var data = stream
    |from()
        .database('influxdb_test')
        .retentionPolicy('autogen')
        .measurement('Processor')
        .groupBy('host')
        .where(lambda: "instance" == '_Total')
    |window()
        .period(2m)
        .every(1m)
    |httpOut('line12')
    |where(lambda: isPresent("Percent_Processor_Time"))
    |last('Percent_Processor_Time')
        .as('Percent_Processor_Time')
    |httpOut('line16')
    |log()

data
    @pyavg()
        .field('Percent_Processor_Time')
        .size(2)
        .as('Moving_Average')
    |log()
    |httpOut('fulloutput')
    |InfluxDBOut()
        .database('influxdb_test')
        .retentionPolicy('autogen')
        .measurement('M_Average')
ashishkaransingh picture ashishkaransingh on 3 Oct 2019

I also tried this in batch, getting same errors in the log.

ashishkaransingh picture ashishkaransingh on 4 Oct 2019

Thank to @nirshar , I lost a week to debug and don't know why my udf doesn't work until your post. Hope this function can be updated soon.

trungkien210493 picture trungkien210493 on 30 Dec 2019
Was this page helpful?
0 / 5 - 0 ratings

Related issues

davidhiendl picture davidhiendl  路  6Comments
R4scal picture R4scal  路  4Comments
mgresser picture mgresser  路  4Comments
cheribral picture cheribral  路  8Comments
evan-ty picture evan-ty  路  3Comments