How to work with events in Flussonic

Working with events in Flussonic for monitoring

Users often ask the question: how to make it so that Flussonic would send a letter when the stream drops.

By turning on the bore you can mutter that it is not clear what a fall is, and so on. There are a lot of questions, because the bitrate of the stream is non-zero, frames are coming, and there will be white noise or a black screen. The flow seems to work, but in fact it does not. But consider the solution of the original problem with the help of the new event system.

The easiest option would be naive, but working. Add to the streamer config:

notify no_video {  sink /etc/flussonic/no_video.lua; } 

in the /etc/flussonic/no_video.lua file /etc/flussonic/no_video.lua write:
')

 for k,event in pairs(events) do --     ,     if event.event == "source_lost" or event.event == "stream_stopped" then --    ,     mail.send({from = "flussonic@streamer1.mycdn", to = "marketing@team.mycdn", subject = "Source lost", body = "source lost on "..event.media}) --        end end 

Prefiltration

It is important not to put a personal email here, because when you run such a code on the production, many letters will have to be raked.

A huge number of events are constantly generated inside the video streamer: for every open / closed session, for every fragment recorded in the archive, etc., so we’ll start optimizing this code by using pre-filtering in the config file:

 notify no_video { sink /etc/flussonic/no_video.lua; only event=source_lost,source_ready,stream_stopped; } 

This is important because filtering occurs in the code that sends the event, and not in the handler, as a result, you can cool off the code and not load one core, and this is very important on multi-core machines.

Configuration

A freshly written lua script will certainly be popular, plus you may want to send letters to one people for one group of threads, and to others for other threads. You can write the script once, and keep its settings in the flusonic configuration:

 notify no_video { sink /etc/flussonic/no_video.lua; only event=source_lost,source_ready,stream_stopped; to marketing@team.mycdn; from flussonic@streamer1.mycdn; } 

and in the script we use the args parameter:

 for k,event in pairs(events) do  if event.event == "source_lost" or event.event == "stream_stopped" then   mail.send({from = args.from, to = args.to, subject = "Source lost", body = "source lost on "..event.media})  end end 

We will slightly correct this script to send not many, but one letter and only if the problem was:

 local body = "" local count = 0 for k,event in pairs(events) do  if event.event == "source_lost" or event.event == "stream_stopped" then   count = count + 1   body = body.."  "..event.media.."\n"  end end if count > 0 then mail.send({from = args.from, to = args.to, subject = "Source lost", body = "source lost on: \n"..body}) end 

Debounce

The script that we have turned out is still far from perfect and it will immediately become clear on a less live production, for example, with a bunch of custom cameras.

Let's find a way to reduce mail traffic and make letters more useful. Having received a message about the fall of the stream, we will not immediately send a letter, but wait for 30 seconds to collect who else has fallen.

To do this, we use the built-in timer mechanism in Flussonic for event handlers. The fact is that Flussonic saves the state of the handler between requests and starts the handler with this state in turn in one stream.

Be careful with this: if you accumulate all the events in a buffer, you can drop the server.

As soon as the first fall event arrives, we will wind up the timer and not send anything until it expires. During this time, we will accumulate information about the fallen flows and then send it all at once.

 if not new_dead_streams then new_dead_streams = {} end local some_stream_died = false for k,evt in pairs(events) do if evt.event == "stream_stopped" or evt.event == "source_lost" then  new_dead_streams[evt.media] = true  some_stream_died = true end end if not notify_timer and some_stream_died then notify_timer = flussonic.timer(30000, "handle_timer", "go") end function handle_timer(arg) local body = "Local time "..flussonic.now().."\n" for name,flag in pairs(new_dead_streams) do  body = body.."  "..name.."\n" end  new_dead_streams = {} mail.send({from = args.from, to = args.to, subject = "Source lost", body = body}) end 

The flussonic.timer function first parameter takes the time to call the function in milliseconds, then the name of the function and then the argument.

It is not possible to transfer a locally declared table, so be careful with this: send a number or string or use a global variable.

To hell with mail, let's be weak

Let's change the mailing list to push in the channel in the slak. You need to go to https://YOURTEAM.slack.com/apps/manage/custom-integrations , from there go to the incoming webhooks

There you will receive a view URL:

 https://hooks.slack.com/services/NB8hv62/ERBAdLVT/VW9teYkRPMp2NMbU 

change mail.send to:

   message = {text= body, username= "flussonic" }  http.post(args.slack_url, {["content-type"] = "application/json"}, json.encode(message)) 

and change the configuration:

 notify no_video { sink /etc/flussonic/no_video.lua; only event=source_lost,source_ready,stream_stopped; slack_url https://hooks.slack.com/services/NB8hv62/ERBAdLVT/VW9teYkRPMp2NMbU; } 

Why do we need args?

It would seem: why pass some parameters to the script, if the script can be corrected?

Firstly, our practice is such that users who even Linux for the first time see us regularly turn to us for help, and editing the lua script for them is simply an unthinkable task: it’s easier to give them a script written by us from the package and put the necessary parameters into the config .

Secondly, the same script can be used with different parameters for different streams. It can be much easier for an experienced user to pass arguments through the config, and not through the heightened conditions in the script itself.

What is there under the hood?

In Flussonic, the system of events was a long time ago, almost since 2012, but the previous implementation was frankly lame. Guided by the name of the module from stdlib, the gen_event module was chosen to implement this mechanism.

In short: never use gen_event , it is not needed anywhere. One of the most important reasons: if a handler fails for gen_event for some reason, it silently silently (or with a small curse in the logs) will be deleted and there will be no restart.

We have changed the internal structure of event handling. Each handler described in the config is launched in a separate gen_server and will be restarted in case of a crash. Each such process handler tries on each message to collect in a pack the maximum number of already committed events (but not more than a limit, about a thousand) and pass them all together to the processor. This approach should reduce the overhead of sending over HTTP or writing to a file.

This alteration made it possible to transfer the filtering of events to the sending process: so if you have a thousand streams on the server, and you only need to keep track of three, then not one core will be involved in filtering, but all 48 or how many you have. This is an essential point for system scalability.

A subtle, but important, nuance is that event handlers know how to monitor overloading and, in principle, even reset events. If things are very bad, then flussonic will not endlessly accumulate events in the queues, but will start throwing them out. Checking for congestion is very nontrivial, because in such cases it is impossible to pour the length of the queue of messages of the receiving process, so we do this through an independent check by the process sending events of our own liveliness.

Conclusion

In this article, I talked a little bit about how:

# make your event handler in Flussonic on lua
# send mail from lua: mail.send # make HTTP request with json body: http.post and json.encode # use timer: flussonic.timer .