How to work with events at Flussonic / Erlivideo company blog

Work with events in Flussonic for monitoring

Users often ask: how to make Flussonic send a message when the stream drops.

Turning on the bore, you can mumble that it is not clear what a fall is, etc. There are a lot of questions, because the bitrate of the stream is non-zero, the frames are on, and there will be white noise or a black screen. The stream seems to work, but essentially not. But consider the solution to the original problem using the new event system.

The easiest option would be naive, but working. In the streamer config, add:

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

in the file /etc/flussonic/no_video.luawe write:

for k,event in pairs(events) do -- события приходят в обработчик пачками, обработаем целиком группу
   if event.event == "source_lost" or event.event == "stream_stopped" then -- отфильтруем только те события, которые нужны
     mail.send({from = "[email protected]", to = "[email protected]", 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 code on production, you will have to erase a lot of letters.

A huge number of events are constantly generated inside the video streamer: for each open / closed session, for each fragment recorded in the archive, etc., so let's start optimizing this code using prefiltering in the config:

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, which is very important on multi-core machines.

Configuration

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

notify no_video {
  sink /etc/flussonic/no_video.lua;
  only event=source_lost,source_ready,stream_stopped;
  to [email protected];
  from [email protected];
}

and in the script we use the parameter args:

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’ll tweak this script a bit to send not a lot, 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 is far from perfect and it will immediately become clear on a less live production for example, with a bunch of user cameras.

Let's find a way to reduce email traffic and make email more useful. After receiving a message about the fall of the stream, we will not immediately send a letter, but wait 30 seconds to collect who else fell.

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

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

As soon as the first event of the fall arrives, we will cock the timer and not send anything until it expires. During this time, we will accumulate information about fallen flows and then we will send everything 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.timerfirst parameter takes the function call time in milliseconds, then the function name and then the argument.

Transferring a locally declared table will fail, so be careful with this: send a number or a string or use a global variable.

To hell with mail, let's slack

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

There you will get the url of the form:

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

change mail.sendto:

    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 so?

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

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

What is there under the hood?

In Flussonic, the event system has been around for a long time, 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 for some reason the handler crashed in gen_event, it will silently silently (or with a slight curse in the logs) be deleted and there will be no restart.

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

This alteration allowed transferring the filtering of events to the sending process: thus, if you have a thousand streams on your server, and you only need to monitor three, then not one core will be filtered, but all 48 or how many you have there. This is an essential point for system scalability.

A subtle but important nuance is that event handlers can monitor overload and, in principle, even reset events. If things are very bad, then fluxonik will not endlessly accumulate events in the queues, but will start throwing them out. Checking for congestion is done very non-trivially, because in such cases it is impossible to fill the length of the message queue of the receiving process, so we do this through an independent check by the process sending events of their own liveliness.

Conclusion

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

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