AzureExporter not working with multiprocessing

[dpgrodriguez]

Describe your environment.
MacOS 10.14.6
Python 3.7.5
opencensus-ext-azure==1.0.4
opencensus-ext-requests==0.7.3

Steps to reproduce.
I have code that I want to monitor the dependency calls to Azure DevOps APIs. Our code is running multiprocessing using the Process class. When the exporter is ran outside of multiprocessing, it sends telemetry to App Insights. When ran inside a multiprocessing Process, it doesn't. I added a callback to print the spandata and it doesn't get called when using Process.

from azure.devops.connection import Connection
from msrest.authentication import BasicAuthentication

from multiprocessing import Process, Pool, Queue
from base_insights import BaseInsights
from opencensus.common.runtime_context import RuntimeContext


class TestInsights:
    def __init__(self):
        self.tracer = BaseInsights.tracer

    def process(self):
       
        procs = []
        organization_url = 'https://dev.azure.com/org'
        credentials = BasicAuthentication('', '')
        
        p1 = Process(target=self.my_loop, args=[organization_url, credentials])
        p1.start()
        p1.join()

    def my_loop(self, organization_url, credentials, parent_span=None):
      
        with self.tracer.span(name='TestLoopProcessThreadingInside'):

            connection = Connection(base_url=organization_url, creds=credentials)
            core_client = connection.clients.get_core_client()

            org = core_client.get_project_collection("test")


TestInsights().process()

BaseInsights:

import os

from opencensus.ext.azure.trace_exporter import AzureExporter
from opencensus.trace import file_exporter

from opencensus.trace import config_integration
from opencensus.trace.samplers import ProbabilitySampler, AlwaysOnSampler
from opencensus.trace.tracer import Tracer

config_integration.trace_integrations(['requests'])


def singleton(cls):
    return cls()


@singleton
class BaseInsights:
    def __init__(self):
        exporter = AzureExporter()
        exporter.add_telemetry_processor(self.callback_function)

        self.tracer = Tracer(exporter=exporter, sampler=AlwaysOnSampler())


    def callback_function(self, envelope):
        print(envelope)

What is the expected behavior?
Span data gets sent to Application Insights

What is the actual behavior?
Span data is not sent to Application Insights

[lzchen]

@dpgrodriguez
I created a more simplified version of your first code snippet by taking out the calls to devops API. The span datas are printed fine and I can see the dependencies in App Insights. I think the process might be exiting too quickly for the exporter to actually export the data. The default export interval is 15seconds. You can try modifying that or setting a delay within your new process.

exporter = AzureExporter(export_interval=5.0)

import time
from multiprocessing import Process
from base_insights import BaseInsights


class TestInsights:
    def __init__(self):
        self.tracer = BaseInsights.tracer

def my_loop():
    
    with BaseInsights.tracer.span(name='TestLoopProcessThreadingInside'):
        print("inside")
        time.sleep(10)

if __name__ == '__main__':
    with BaseInsights.tracer.span(name='TestOutside'):
        print("outside")
        time.sleep(10)
    p1 = Process(target=my_loop)
    p1.start()
    p1.join()

[dpgrodriguez]

Hi Leighton,

I tried your suggestion but still cannot send span data from inside multiprocess. I used your exact code and this is what was displayed:

outside
{...redacted envelope data from outside}
inside

The only span data generated was from "outside".

I also tried removing the parent span from __main__:

if __name__ == '__main__':
#    with BaseInsights.tracer.span(name='TestOutside'):
    print("outside")
    time.sleep(10)
    p1 = Process(target=my_loop)
    p1.start()
    p1.join()

with these results:

outside
inside

The data from inside multiprocessing also wasn't sent to Applicaiton Insights

These are the versions that I'm using:
opencensus==0.7.10
opencensus-context==0.1.1
opencensus-ext-azure==1.0.4
opencensus-ext-requests==0.7.3
opencensus-ext-threading==0.1.2

[lzchen]

Did you set the export_interval of the exporter? The whole execution from start to finish should take around 20 seconds.

[dpgrodriguez]

Yes, tried setting it to 5 and 1.

[dpgrodriguez]

Any other recommendations?

[lzchen]

Perhaps it's the way you are setting up BaseInsights? I simplified it even further. Try to see if it works for you.

import time
from multiprocessing import Process
from opencensus.trace.samplers import AlwaysOnSampler
from opencensus.trace.tracer import Tracer
from opencensus.ext.azure.trace_exporter import AzureExporter

exporter = AzureExporter(export_interval=5)
tracer = Tracer(exporter=exporter, sampler=AlwaysOnSampler())

def callback_function(envelope):
    print(envelope)

exporter.add_telemetry_processor(callback_function)

def my_loop():
    
    with tracer.span(name='TestLoopProcessThreadingInside'):
        print("inside")
        time.sleep(10)

def main():
    with tracer.span(name='TestOutside'):
        print("outside")
        time.sleep(10)
        p1 = Process(target=my_loop)
        p1.start()
        p1.join()

if __name__ == '__main__':
    main()

[dpgrodriguez]

I tried running this code on 4 different environments.

CentOS 7.8 - Python 2.7.5
CentOS 7.8 - Python 3.6.8
macOS 10.14.6 - Python 2.7.16
macOS 10.14.6 - Python 3.7.5

I got the same results. Only INPROC: TestOutside is being logged in Application Insights.

[dpgrodriguez]

Is there a way to show lower level debug messages?

[lzchen]

@dpgrodriguez
I did some debugging and it looks like the multiprocessing module behaves a little differently on Windows then macOC/linux. On Windows, when a new process is created, all threads are copied over into the new address space (including the worker that sends telemetry to Azure Monitor), however when I ran the same code in Linux, only the calling thread was copied. The multiprocessing module uses os.fork() underneath for POSIX systems (which includes linux), and from the documentation of fork: A process shall be created with a single thread. If a multi-threaded process calls fork(), the new process shall contain a replica of the calling thread and its entire address space, possibly including the states of mutexes and other resources. Whereas, Windows will spin up an entirely new Process, and tell it to load all the modules again (which in turn creates another AzureExporter and worker).

So the only way to get this working for your OS is to duplicate the initialization logic of your AzureExporter and Tracer into the function that is run in the spawned Process.

[dpgrodriguez]

That's some interesting findings. I did some digging as well.

In BaseExporter (opencensus.ext.azure.common.exporter.py), it looks like the items are actually being passed into the export method and into the queue even inside the multiprocess Process.

    def export(self, items):
        self._queue.puts(items, block=False)  # pragma: NO COVER

However, in the Worker's run method, as it tries to get the batch from the queue, it cannot find the items anymore:

class Worker(threading.Thread):
    daemon = True
...
...

        while True:
            batch = src.gets(dst.max_batch_size, dst.export_interval)

Any thoughts on this? Could it be that the threading Queue is not playing well with multiprocessing?

[lzchen]

@dpgrodriguez
Yes there is nothing in the queue because the Worker that fills them up doesn't exist.

[dpgrodriguez]

The worker is actually present but doesn't see the contents of the threading queue because it can't access it. I did some quick PoC of using the multiprocess Queue and was able to send logs to analytics.

from multiprocessing import Queue as MP_Queue
import jsonpickle

...redacted...

    # queue, or shared workers among queues (e.g. queue for traces, queue
    # for logs).
    def export(self, items):
        json_items = jsonpickle.encode(items)

        # Put items on both multiprocessing and threading queues
        self._mp_queue.put(json_items)
        self._queue.puts(items, block=False)  # pragma: NO COVER


... redacted ...

    def run(self):  # pragma: NO COVER
        # Indicate that this thread is an exporter thread.
        # Used to suppress tracking of requests in this thread.
        execution_context.set_is_exporter(True)
        src = self.src
        dst = self.dst
        while True:
            batch = src.gets(dst.max_batch_size, dst.export_interval)

            # Check if batch results in empty tuple and check multiprocessing queue for contents
            if batch == ():
                try:
                    json_items = dst._mp_queue.get()
                    batch = tuple(jsonpickle.decode(json_items))
                except Exception as e:
                    pass

            if batch and isinstance(batch[-1], QueueEvent):

... redacted ...

This works and sends the multiprocessed SpanData into insights.