implemented http/xml/xsl RTMP stats

TODO

@@ -1,15 +1,8 @@
-- implement json http module for accessing stats
-  * memory stats (shared bufs num/bytes, private pools
-  * global (connections, IO bytes, bw)
-  * live (stream name, flags, clients, bytes, bw, drops)
-
-- add global & per-room in/out byte counters
-
 - add per-client audio/video bias
 
 - write complete WIKI docs
 
-- rename netcall to upstream
+- global rename: get rid of '_module' suffixes in files && rename netcall to upstream
 
 - add support for flv file streaming 
   (NGINX supports streaming flvs via HTTP

config

@@ -11,6 +11,12 @@ CORE_MODULES="$CORE_MODULES
                 ngx_rtmp_notify_module                      \
                 "
 
+
+HTTP_MODULES="$HTTP_MODULES                                 \
+                ngx_rtmp_stat_module                        \
+                "
+
+
 NGX_ADDON_SRCS="$NGX_ADDON_SRCS                             \
                 $ngx_addon_dir/ngx_rtmp.c                   \
                 $ngx_addon_dir/ngx_rtmp_amf.c               \
@@ -25,5 +31,7 @@ NGX_ADDON_SRCS="$NGX_ADDON_SRCS                             \
                 $ngx_addon_dir/ngx_rtmp_record_module.c     \
                 $ngx_addon_dir/ngx_rtmp_netcall_module.c    \
                 $ngx_addon_dir/ngx_rtmp_notify_module.c     \
+                $ngx_addon_dir/ngx_rtmp_stat_module.c       \
+                $ngx_addon_dir/ngx_rtmp_bandwidth.c         \
                 "
 CFLAGS="$CFLAGS -I$ngx_addon_dir"

ngx_rtmp.h

@@ -13,6 +13,7 @@
 #include <ngx_event_connect.h>
 
 #include "ngx_rtmp_amf.h"
+#include "ngx_rtmp_bandwidth.h"
 
 
 typedef struct {
@@ -261,6 +262,10 @@ typedef struct {
 } ngx_rtmp_core_main_conf_t;
 
 
+/* global main conf for stats */
+extern ngx_rtmp_core_main_conf_t   *ngx_rtmp_core_main_conf;
+
+
 typedef struct ngx_rtmp_core_srv_conf_s {
     ngx_array_t             applications; /* ngx_rtmp_core_app_conf_t */
 
@@ -467,8 +472,12 @@ ngx_rtmp_get_video_frame_type(ngx_chain_t *in)
 }
 
 
-extern ngx_uint_t    ngx_rtmp_max_module;
-extern ngx_module_t  ngx_rtmp_core_module;
+extern ngx_rtmp_bandwidth_t                 ngx_rtmp_bw_out;
+extern ngx_rtmp_bandwidth_t                 ngx_rtmp_bw_in;
+
+
+extern ngx_uint_t                           ngx_rtmp_max_module;
+extern ngx_module_t                         ngx_rtmp_core_module;
 
 
 #endif /* _NGX_RTMP_H_INCLUDED_ */

ngx_rtmp_bandwidth.c

@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2012 Roman Arutyunyan
+ */
+
+
+#include "ngx_rtmp_bandwidth.h"
+
+
+void 
+ngx_rtmp_update_bandwidth(ngx_rtmp_bandwidth_t *bw, uint32_t bytes)
+{
+    if (ngx_cached_time->sec > bw->intl_end) {
+        bw->bandwidth = ngx_cached_time->sec > 
+            bw->intl_end + NGX_RTMP_BANDWIDTH_INTERVAL
+            ? 0
+            : bw->intl_bytes / NGX_RTMP_BANDWIDTH_INTERVAL;
+        bw->intl_bytes = 0;
+        bw->intl_end = ngx_cached_time->sec + NGX_RTMP_BANDWIDTH_INTERVAL;
+    }
+
+    bw->bytes += bytes;
+    bw->intl_bytes += bytes;
+}

ngx_rtmp_bandwidth.h

@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2012 Roman Arutyunyan
+ */
+
+
+#ifndef _NGX_RTMP_BANDWIDTH_H_INCLUDED_
+#define _NGX_RTMP_BANDWIDTH_H_INCLUDED_
+
+
+#include <ngx_core.h>
+
+
+/* Bandwidth update interval in seconds */
+#define NGX_RTMP_BANDWIDTH_INTERVAL     60
+
+
+typedef struct {
+    uint64_t            bytes;
+    uint64_t            bandwidth;      /* bytes/sec */
+
+    time_t              intl_end;
+    uint64_t            intl_bytes;
+} ngx_rtmp_bandwidth_t;
+
+
+void ngx_rtmp_update_bandwidth(ngx_rtmp_bandwidth_t *bw, uint32_t bytes);
+
+
+#endif /* _NGX_RTMP_BANDWIDTH_H_INCLUDED_ */
+

ngx_rtmp_cmd_module.c

@@ -70,7 +70,7 @@ ngx_rtmp_cmd_connect_init(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
           v.app, sizeof(v.app) },
 
         { NGX_RTMP_AMF_STRING, 
-          ngx_string("flashver"),
+          ngx_string("flashVer"),
           v.flashver, sizeof(v.flashver) },
 
         { NGX_RTMP_AMF_STRING,

ngx_rtmp_core_module.c

@@ -25,6 +25,9 @@ static char *ngx_rtmp_core_application(ngx_conf_t *cf, ngx_command_t *cmd,
     void *conf);
 
 
+ngx_rtmp_core_main_conf_t      *ngx_rtmp_core_main_conf;
+
+
 static ngx_conf_deprecated_t  ngx_conf_deprecated_so_keepalive = {
     ngx_conf_deprecated, "so_keepalive", 
     "so_keepalive\" parameter of the \"listen"
@@ -160,6 +163,8 @@ ngx_rtmp_core_create_main_conf(ngx_conf_t *cf)
         return NULL;
     }
 
+    ngx_rtmp_core_main_conf = cmcf;
+
     if (ngx_array_init(&cmcf->servers, cf->pool, 4,
                        sizeof(ngx_rtmp_core_srv_conf_t *))
         != NGX_OK)

ngx_rtmp_handler.c

@@ -26,6 +26,10 @@ static ngx_int_t ngx_rtmp_receive_message(ngx_rtmp_session_t *s,
 static ngx_int_t ngx_rtmp_finalize_set_chunk_size(ngx_rtmp_session_t *s);
 
 
+ngx_rtmp_bandwidth_t        ngx_rtmp_bw_out;
+ngx_rtmp_bandwidth_t        ngx_rtmp_bw_in;
+
+
 #ifdef NGX_DEBUG
 char*
 ngx_rtmp_message_type(uint8_t type) {
@@ -570,6 +574,7 @@ ngx_rtmp_recv(ngx_event_t *rev)
                 return;
             }
 
+            ngx_rtmp_update_bandwidth(&ngx_rtmp_bw_in, n);
             b->last += n;
             s->in_bytes += n;
 
@@ -825,6 +830,7 @@ ngx_rtmp_send(ngx_event_t *wev)
             return;
         }
 
+        ngx_rtmp_update_bandwidth(&ngx_rtmp_bw_out, n);
         s->out_bpos += n;
         if (s->out_bpos == s->out_chain->buf->last) {
             s->out_chain = s->out_chain->next;

ngx_rtmp_live_module.c

@@ -3,9 +3,7 @@
  */
 
 
-#include <ngx_config.h>
-#include <ngx_core.h>
-#include "ngx_rtmp.h"
+#include "ngx_rtmp_live_module.h"
 #include "ngx_rtmp_cmd_module.h"
 
 
@@ -19,51 +17,13 @@ static ngx_rtmp_delete_stream_pt        next_delete_stream;
 #define NGX_RTMP_LIVE_CSID_VIDEO        7
 #define NGX_RTMP_LIVE_MSID              1
 
-/* session flags */
-#define NGX_RTMP_LIVE_PUBLISHING        0x01
-
 
 static ngx_int_t ngx_rtmp_live_postconfiguration(ngx_conf_t *cf);
 static void * ngx_rtmp_live_create_app_conf(ngx_conf_t *cf);
 static char * ngx_rtmp_live_merge_app_conf(ngx_conf_t *cf, 
         void *parent, void *child);
 
 
-typedef struct ngx_rtmp_live_ctx_s ngx_rtmp_live_ctx_t;
-typedef struct ngx_rtmp_live_stream_s ngx_rtmp_live_stream_t;
-
-
-struct ngx_rtmp_live_ctx_s {
-    ngx_rtmp_session_t                 *session;
-    ngx_rtmp_live_stream_t             *stream;
-    ngx_rtmp_live_ctx_t                *next;
-    ngx_uint_t                          flags;
-    ngx_uint_t                          msg_mask;
-    uint32_t                            csid;
-    uint32_t                            next_push;
-    uint32_t                            last_audio;
-    uint32_t                            last_video;
-};
-
-
-struct ngx_rtmp_live_stream_s {
-    u_char                              name[256];
-    ngx_rtmp_live_stream_t             *next;
-    ngx_rtmp_live_ctx_t                *ctx;
-    ngx_uint_t                          flags;
-};
-
-
-typedef struct {
-    ngx_int_t                           nbuckets;
-    ngx_rtmp_live_stream_t            **streams;
-    ngx_flag_t                          live;
-    ngx_msec_t                          buflen;
-    ngx_pool_t                         *pool;
-    ngx_rtmp_live_stream_t             *free_streams;
-} ngx_rtmp_live_app_conf_t;
-
-
 #define NGX_RTMP_LIVE_TIME_ABSOLUTE     0x01
 #define NGX_RTMP_LIVE_TIME_RELATIVE     0x02
 
@@ -331,6 +291,7 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
     ngx_rtmp_session_t             *ss;
     ngx_rtmp_header_t               ch, lh;
     ngx_uint_t                      prio, peer_prio;
+    ngx_uint_t                      peers, dropped_peers;
 
     c = s->connection;
     lacf = ngx_rtmp_get_module_app_conf(s, ngx_rtmp_live_module);
@@ -387,11 +348,15 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
     out = ngx_rtmp_append_shared_bufs(cscf, NULL, in);
     ngx_rtmp_prepare_message(s, &ch, &lh, out);
 
+    peers = 0;
+    dropped_peers = 0;
+
     /* broadcast to all subscribers */
     for (pctx = ctx->stream->ctx; pctx; pctx = pctx->next) {
         if (pctx == ctx) {
             continue;
         }
+        ++peers;
         ss = pctx->session;
 
         /* send absolute frame */
@@ -411,14 +376,22 @@ ngx_rtmp_live_av(ngx_rtmp_session_t *s, ngx_rtmp_header_t *h,
 
         /* push buffered data */
         peer_prio = prio;
+        /*
         if (lacf->buflen && h->timestamp >= pctx->next_push) {
             peer_prio = 0;
             pctx->next_push = h->timestamp + lacf->buflen;
+        }*/
+        if (ngx_rtmp_send_message(ss, out, peer_prio) != NGX_OK) {
+            ++pctx->dropped;
+            ++dropped_peers;
         }
-        ngx_rtmp_send_message(ss, out, peer_prio);
     }
     ngx_rtmp_free_shared_chain(cscf, out);
 
+    ngx_rtmp_update_bandwidth(&ctx->stream->bw_in, h->mlen);
+    ngx_rtmp_update_bandwidth(&ctx->stream->bw_out, 
+            h->mlen * (peers - dropped_peers));
+
     return NGX_OK;
 }
 

ngx_rtmp_live_module.h

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2012 Roman Arutyunyan
+ */
+
+
+#ifndef _NGX_RTMP_LIVE_H_INCLUDED_
+#define _NGX_RTMP_LIVE_H_INCLUDED_
+
+
+#include "ngx_rtmp.h"
+#include "ngx_rtmp_bandwidth.h"
+
+
+/* session flags */
+#define NGX_RTMP_LIVE_PUBLISHING        0x01
+
+
+typedef struct ngx_rtmp_live_ctx_s ngx_rtmp_live_ctx_t;
+typedef struct ngx_rtmp_live_stream_s ngx_rtmp_live_stream_t;
+
+
+struct ngx_rtmp_live_ctx_s {
+    ngx_rtmp_session_t                 *session;
+    ngx_rtmp_live_stream_t             *stream;
+    ngx_rtmp_live_ctx_t                *next;
+    ngx_uint_t                          flags;
+    ngx_uint_t                          msg_mask;
+    ngx_uint_t                          dropped;
+    uint32_t                            csid;
+    uint32_t                            next_push;
+    uint32_t                            last_audio;
+    uint32_t                            last_video;
+};
+
+
+struct ngx_rtmp_live_stream_s {
+    u_char                              name[256];
+    ngx_rtmp_live_stream_t             *next;
+    ngx_rtmp_live_ctx_t                *ctx;
+    ngx_uint_t                          flags;
+    ngx_rtmp_bandwidth_t                bw_in;
+    ngx_rtmp_bandwidth_t                bw_out;
+};
+
+
+typedef struct {
+    ngx_int_t                           nbuckets;
+    ngx_rtmp_live_stream_t            **streams;
+    ngx_flag_t                          live;
+    ngx_msec_t                          buflen;
+    ngx_pool_t                         *pool;
+    ngx_rtmp_live_stream_t             *free_streams;
+} ngx_rtmp_live_app_conf_t;
+
+
+extern ngx_module_t  ngx_rtmp_live_module;
+
+
+#endif /* _NGX_RTMP_LIVE_H_INCLUDED_ */