Prefix all the Futatabi Prometheus metrics by futatabi_ instead of nageru_.

[nageru] / futatabi / main.cpp

diff --git a/futatabi/main.cpp b/futatabi/main.cpp
index 5f262353eed6a55246286a557af0456ae2608528..52cdb80664379e5014fb0f5ea673234e087f14b8 100644 (file)
--- a/futatabi/main.cpp
+++ b/futatabi/main.cpp
@@ -34,6 +34,7 @@ extern "C" {
 #include "shared/post_to_main_thread.h"
 #include "shared/ref_counted_gl_sync.h"
 #include "shared/timebase.h"
+#include "shared/metrics.h"
 #include "ui_mainwindow.h"
 #include "vaapi_jpeg_decoder.h"
 
@@ -69,6 +70,8 @@ mutex frame_mu;
 vector<FrameOnDisk> frames[MAX_STREAMS];  // Under frame_mu.
 vector<string> frame_filenames;  // Under frame_mu.
 
+atomic<int64_t> metric_received_frames[MAX_STREAMS]{{0}};
+
 namespace {
 
 FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t size, DB *db)
@@ -183,7 +186,7 @@ FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t
 HTTPD *global_httpd;
 
 void load_existing_frames();
-int record_thread_func();
+void record_thread_func();
 
 int main(int argc, char **argv)
 {
@@ -200,16 +203,15 @@ int main(int argc, char **argv)
 
        string frame_dir = global_flags.working_directory + "/frames";
 
-       struct stat st;
-       if (stat(frame_dir.c_str(), &st) == -1) {
+       if (mkdir(frame_dir.c_str(), 0777) == 0) {
                fprintf(stderr, "%s does not exist, creating it.\n", frame_dir.c_str());
-               if (mkdir(frame_dir.c_str(), 0777) == -1) {
-                       perror(global_flags.working_directory.c_str());
-                       exit(1);
-               }
+       } else if (errno != EEXIST) {
+               perror(global_flags.working_directory.c_str());
+               exit(1);
        }
 
        avformat_network_init();
+       global_metrics.set_prefix("futatabi");
        global_httpd = new HTTPD;
 
        QCoreApplication::setAttribute(Qt::AA_ShareOpenGLContexts, true);
@@ -282,7 +284,7 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
        if (!all_frames.empty()) {
                // We already had this cached in the database, so no need to look in the file.
                for (const DB::FrameOnDiskAndStreamIdx &frame : all_frames) {
-                       if (frame.stream_idx >= 0 && frame.stream_idx < MAX_STREAMS) {
+                       if (frame.stream_idx < MAX_STREAMS) {
                                frames[frame.stream_idx].push_back(frame.frame);
                                start_pts = max(start_pts, frame.frame.pts);
                        }
@@ -344,6 +346,10 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
                FrameOnDisk frame;
                frame.pts = hdr.pts();
                frame.offset = ftell(fp);
+               if (frame.offset == -1) {
+                       fprintf(stderr, "WARNING: %s: ftell() failed (%s).\n", filename, strerror(errno));
+                       break;
+               }
                frame.filename_idx = filename_idx;
                frame.size = hdr.file_size();
 
@@ -364,9 +370,14 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
                        filename, skipped_bytes);
        }
 
-       size_t size = ftell(fp);
+       off_t size = ftell(fp);
        fclose(fp);
 
+       if (size == -1) {
+               fprintf(stderr, "WARNING: %s: ftell() failed (%s).\n", filename, strerror(errno));
+               return;
+       }
+
        db->store_frame_file(basename, size, all_frames);
 }
 
@@ -444,11 +455,20 @@ void load_existing_frames()
        db.clean_unused_frame_files(frame_basenames);
 }
 
-int record_thread_func()
+void record_thread_func()
 {
+       for (unsigned i = 0; i < MAX_STREAMS; ++i) {
+               global_metrics.add("received_frames", {{ "stream", to_string(i) }}, &metric_received_frames[i]);
+       }
+
+       if (global_flags.stream_source.empty() || global_flags.stream_source == "/dev/null") {
+               // Save the user from some repetitive messages.
+               return;
+       }
+
        pthread_setname_np(pthread_self(), "ReceiveFrames");
 
-       int64_t pts_offset;
+       int64_t pts_offset = 0;  // Needs to be initialized due to a spurious GCC warning.
        DB db(global_flags.working_directory + "/futatabi.db");
 
        while (!should_quit.load()) {
@@ -474,6 +494,11 @@ int record_thread_func()
                        if (av_read_frame(format_ctx.get(), &pkt) != 0) {
                                break;
                        }
+                       if (pkt.stream_index >= MAX_STREAMS) {
+                               continue;
+                       }
+
+                       ++metric_received_frames[pkt.stream_index];
 
                        // Convert pts to our own timebase.
                        AVRational stream_timebase = format_ctx->streams[pkt.stream_index]->time_base;
@@ -505,5 +530,4 @@ int record_thread_func()
 
                start_pts = last_pts + TIMEBASE;
        }
-       return 0;
 }