Merge pull request #14 from kaltura/transcoder-log-json

Move transcoder log to Json format

Author: david-winder-kaltura

transcoder/main.c

@@ -166,6 +166,7 @@ int set_resource_limits()
     }
     return 0;
 }
+extern char *configString;
 
 int main(int argc, char **argv)
 {
@@ -176,8 +177,6 @@ int main(int argc, char **argv)
 
     log_init(AV_LOG_DEBUG);
 
-    LOGGER(CATEGORY_DEFAULT,AV_LOG_INFO,"Version: %s", APPLICATION_VERSION)
-
     signal(SIGINT, intHandler);
     signal(SIGPIPE, pipeHandler);
 
@@ -191,6 +190,15 @@ int main(int argc, char **argv)
         set_log_level(logLevel);
     }
 
+    bool logOutputJson;
+    if (JSON_OK==json_get_bool(GetConfig(),"logger.logOutputJson",false,&logOutputJson)) {
+        set_log_output_json(logOutputJson);
+    }
+
+    LOGGER(CATEGORY_DEFAULT,AV_LOG_INFO,"Version: %s", APPLICATION_VERSION)
+
+    LOGGER(CATEGORY_DEFAULT,AV_LOG_INFO,"Parsed configuration successfully: %s",configString);
+
     avformat_network_init();
 
     if (start() < 0) {

transcoder/tests/config_v.json

@@ -23,7 +23,10 @@
     },
     "logger": {
         "logLevels": ["DEBUG","VERBOSE","INFO","WARN","ERROR","FATAL","PANIC"],
-        "logLevel": "DEBUG"
+        "logLevel": "DEBUG",
+        "logOutputJson": true,
+        "channelId": "1_2345678",
+        "contextId": "ABCDE1234"
     },
     "kmp": {
         "listenPort": 16543,

transcoder/utils/config.c

@@ -71,6 +71,6 @@ int LoadConfig(int argc, char **argv)
         LOGGER(CATEGORY_DEFAULT,AV_LOG_FATAL,"Failed parsing configurtion! %s (%s)",configString,error);
         return -1;
     }
-    LOGGER(CATEGORY_DEFAULT,AV_LOG_INFO,"Parsed configuration successfully: %s",configString);
+
     return 0;
 }

transcoder/utils/logger.c

@@ -11,6 +11,8 @@
 
 static int logLevel =AV_LOG_VERBOSE;
 
+static bool logOutputJson = false;
+
 const char* getLevelStr(int level) {
     switch(level){
         case AV_LOG_PANIC: return "PANIC";
@@ -48,7 +50,12 @@ int parseLoglevel(const char* loglevel)
 
 pthread_mutex_t logger_locker;
 
-char logger_id[256] = {0};
+char context_id[256] = {0};
+char channel_id[256] = {0};
+
+static int stringify_message(FILE *out, const char *fmt, va_list args);
+
+void logger_unit_test();
 
 void logger2(const char* category,const char* subcategory,int level,const char *fmt, bool newLine, va_list args)
 {
@@ -66,14 +73,22 @@ void logger2(const char* category,const char* subcategory,int level,const char *
 
     FILE* out=stdout;
 
-    fprintf( out, "%s.%03lld %s:%s %s |%s| [%p] ",buf,( (now % 1000000)/1000 ),category,subcategory!=NULL ? subcategory : "", levelStr,logger_id,pthread_self());
-    if (args!=NULL) {
-        vfprintf( out, fmt, args );
+    if(logOutputJson) {
+        fprintf( out, "{\"time\": \"%s.%03ld\", \"channelId\": \"%s\", \"category\": \"%s:%s\", \"logLevel\": \"%s\","
+            "\"contextId\": \"%s\", \"pthread\":\"%lx\", \"log\": ",buf,( (now % 1000000)/1000 ), channel_id, category,
+             subcategory!=NULL ? subcategory : "", levelStr,context_id,pthread_self());
+        stringify_message(out, fmt, args);
+        fprintf( out, "}\n" );
     } else {
-        fprintf(out,"%s",fmt);
-    }
-    if (newLine) {
-        fprintf( out, "\n" );
+        fprintf( out, "%s.%03ld %s:%s %s |%s| [%s] [%lx] ",buf,( (now % 1000000)/1000 ),category,subcategory!=NULL ? subcategory : "", levelStr,context_id,channel_id,pthread_self());
+         if (args!=NULL) {
+            vfprintf( out, fmt, args );
+        } else {
+            fwrite(fmt, strlen(fmt), 1, out);
+        }
+        if (newLine) {
+            fprintf( out, "\n" );
+        }
     }
     fflush(out);
     pthread_mutex_unlock(&logger_locker); // unlock once you are done
@@ -141,17 +156,149 @@ void log_init(int level)
 
 void set_log_level(const char* loglevel) {
     logLevel=parseLoglevel(loglevel);
-    if(!*logger_id) {
-        json_get_string(GetConfig(),"logger.id","\0",logger_id,sizeof(logger_id));
+    if(!*context_id) {
+        json_get_string(GetConfig(),"logger.contextId","\0",context_id,sizeof(context_id));
+    }
+    if (!*channel_id) {
+        json_get_string(GetConfig(),"logger.channelId","\0",channel_id,sizeof(channel_id));
     }
 }
 
 int get_log_level(const char* category)
 {
     return logLevel;
 }
+
+void set_log_output_json(bool val)
+{
+    logOutputJson = val;
+}
+
+bool get_log_output_json()
+{
+    return logOutputJson;
+}
+
 void loggerFlush()
 {
     fflush(stderr);
     fflush(stdout);
 }
+
+/*
+    json logger utils
+*/
+
+#define FWRITE_STR(x) fwrite(x,1,sizeof(x)-1,out)
+
+/*
+    escapes string according to https://tc39.es/ecma262/multipage/structured-data.html#sec-json.stringify
+*/
+const char *hex = "0123456789ABCDEF";
+static size_t json_escape(FILE *out, const char *str) {
+    const char *start = str;
+	while (*str) {
+		char chr = *str;
+
+		if (chr == '"' || chr == '\\' || chr == '/') {
+			FWRITE_STR("\\");
+			putc(chr,out);
+		} else if (chr == '\b')	{
+		    FWRITE_STR("\\b");
+		} else if (chr == '\f')	{
+		    FWRITE_STR("\\f");
+		} else if (chr == '\n')	{
+		    FWRITE_STR("\\n");
+		} else if (chr == '\r')	{
+		    FWRITE_STR("\\r");
+		} else if (chr == '\t')	{
+		    FWRITE_STR("\\t");
+		} else if (!isprint(chr)) {
+		    FWRITE_STR("\\u");
+			for (int i = 0; i < 4; i++)	{
+				int value = (chr >> 12) & 0xf;
+				if (value >= 0 && value <= 15)
+					putc(hex[value],out);
+				chr <<= 4;
+			}
+		} else {
+			putc(chr,out);
+		}
+        str++;
+	}
+	return str - start;
+}
+
+typedef struct {
+  char *buf;
+  size_t size;
+  FILE *fp;
+} mem_stream_t;
+
+static mem_stream_t aux_mem_stream = {.buf = NULL,.size = 0,.fp = NULL};
+
+static int open_mem_stream(mem_stream_t *stream) {
+    if(!stream->fp){
+        stream->fp = open_memstream(&stream->buf,&stream->size);
+    }
+    if(!stream->fp) {
+        fprintf(stderr,"ERROR: could not open_memstream - os error code %d", errno);
+        return -1;
+    }
+    return 0;
+}
+
+/*
+    stringify_message. prints log message as a json stringified string
+    NOTE: not thread safe! must be called from within a lock.
+*/
+static int stringify_message(FILE *out, const char *fmt, va_list args) {
+      int ret = -1;
+      FWRITE_STR("\"");
+      if (args!=NULL) {
+        if((ret = open_mem_stream(&aux_mem_stream))){
+           goto error;
+        }
+        if((ret = fseek(aux_mem_stream.fp, 0, SEEK_SET))) {
+            fprintf(stderr,"ERROR: could not fseek %d - os error code %d", ret, errno);
+            goto error;
+         }
+         if( (ret = vfprintf(aux_mem_stream.fp, fmt, args )) > 0){
+            fflush(aux_mem_stream.fp);
+            aux_mem_stream.buf[ret] = '\0';
+            ret = json_escape(out, aux_mem_stream.buf);
+         } else {
+            fprintf(stderr,"WARNING: vfprintf error %d. os error code %d", ret, errno);
+         }
+      } else {
+         ret = json_escape(out, fmt);
+      }
+error:
+      FWRITE_STR("\"");
+      return ret;
+}
+
+/*
+    runtime test
+*/
+void logger_unit_test() {
+      const char *special = "\a\b\t\n\r\f\"\\";
+      char table [255+sizeof(special)-1],
+            *walker = table;
+      memcpy(walker,special,strlen(special));
+      walker += strlen(special);
+      for(int i = 1; i < 256; i++) {
+          *walker++ = (char)i;
+      }
+
+      srand((unsigned) time(NULL));
+      for (int i = 0; i < 1024; i++) {
+          int a = rand() % sizeof(table), b =  rand() % sizeof(table);
+          int temp = table[a];
+          table[a] = table[b];
+          table[b] = temp;
+
+          LOGGER(CATEGORY_DEFAULT,AV_LOG_WARNING,"%s", table);
+      }
+      exit(0);
+}