sipwise
/
rtpengine
mirror of https://github.com/sipwise/rtpengine.git
1
0
Fork 0
Code Issues Releases Wiki Activity

MT#64939 add S3/GCS storage info to DB

Browse Source 
Change-Id: Ibe1757dc9818c8e61383718bf58914ba2403da62
pull/2101/head
Richard Fuchs 3 months ago
parent 771d24bcd4
commit bac85c7d68
7 changed files with 107 additions and 2 deletions
Show Stats Download Patch File Download Diff File

60
recording-daemon/db.c
Unescape View File

@ -32,9 +32,9 @@ CREATE TABLE `recording_streams` (
`full_filename` varchar(250) NOT NULL,
`file_format` varchar(10) NOT NULL,
`output_type` enum('mixed','single') NOT NULL,
`stream_id` int(10) unsigned NOT NULL,
`start_time` timestamp NOT NULL DEFAULT current_timestamp(),
`end_time` datetime DEFAULT NULL,
`stream_id` int(10) unsigned NOT NULL,
`sample_rate` int(10) unsigned NOT NULL DEFAULT '0',
`channels` int(10) unsigned NOT NULL DEFAULT '0',
`ssrc` int(10) unsigned NOT NULL,
@ -44,6 +44,12 @@ CREATE TABLE `recording_streams` (
`stream` longblob NOT NULL DEFAULT '',
`transcript_status` enum('none','pending','done') NOT NULL DEFAULT 'none',
`transcript` text NOT NULL DEFAULT '',
`storage_file` tinyint(1) NOT NULL DEFAULT 1,
`storage_db` tinyint(1) NOT NULL DEFAULT 0,
`storage_s3` tinyint(1) NOT NULL DEFAULT 0,
`storage_gcs` tinyint(1) NOT NULL DEFAULT 0,
`s3_uri` varchar(255) NOT NULL DEFAULT '',
`gcs_uri` varchar(255) NOT NULL DEFAULT '',
PRIMARY KEY (`id`),
KEY `call` (`call`),
KEY `transcript_status_call_idx` (`transcript_status`,`call`),
@ -74,6 +80,9 @@ struct db_conn {
*stm_close_stream,
*stm_delete_stream,
*stm_config_stream,
*stm_stream_rm_file,
*stm_stream_s3,
*stm_stream_gcs,
*stm_insert_metadata;
};
typedef struct db_conn db_conn_t;
@ -111,6 +120,9 @@ static void reset_conn(db_conn_t *dbc) {
my_stmt_close(&dbc->stm_delete_stream);
my_stmt_close(&dbc->stm_config_stream);
my_stmt_close(&dbc->stm_insert_metadata);
my_stmt_close(&dbc->stm_stream_s3);
my_stmt_close(&dbc->stm_stream_gcs);
my_stmt_close(&dbc->stm_stream_rm_file);
mysql_close(dbc->mysql_conn);
}
@ -188,7 +200,7 @@ static db_conn_t *check_conn(void) {
goto err;
if ((output_storage & OUTPUT_STORAGE_DB)) {
if (prep(dbc, &dbc->stm_close_stream, "update recording_streams set " \
"end_timestamp = ?, stream = ? where id = ?"))
"end_timestamp = ?, stream = ?, storage_db = 1 where id = ?"))
goto err;
}
else {
@ -200,6 +212,12 @@ static db_conn_t *check_conn(void) {
goto err;
if (prep(dbc, &dbc->stm_config_stream, "update recording_streams set channels = ?, sample_rate = ? where id = ?"))
goto err;
if (prep(dbc, &dbc->stm_stream_rm_file, "update recording_streams set storage_file = 0 where id = ?"))
goto err;
if (prep(dbc, &dbc->stm_stream_s3, "update recording_streams set storage_s3 = 1, s3_uri = ? where id = ?"))
goto err;
if (prep(dbc, &dbc->stm_stream_gcs, "update recording_streams set storage_gcs = 1, gcs_uri = ? where id = ?"))
goto err;
if (prep(dbc, &dbc->stm_insert_metadata, "insert into recording_metakeys (`call`, `key`, `value`) values " \
"(?,?,?)"))
goto err;
@ -514,3 +532,41 @@ void db_config_stream(output_t *op) {
execute_wrap(stm_config_stream, b, NULL);
}
void db_set_s3(unsigned long long db_id, const char *uri) {
if (!db_wanted())
return;
if (db_id == 0)
return;
MYSQL_BIND b[2];
my_cstr(&b[0], uri);
my_ull(&b[1], &db_id);
execute_wrap(stm_stream_s3, b, NULL);
}
void db_set_gcs(unsigned long long db_id, const char *uri) {
if (!db_wanted())
return;
if (db_id == 0)
return;
MYSQL_BIND b[2];
my_cstr(&b[0], uri);
my_ull(&b[1], &db_id);
execute_wrap(stm_stream_gcs, b, NULL);
}
void db_rm_file(output_t *op) {
if (!db_wanted())
return;
if (op->db_id == 0)
return;
MYSQL_BIND b[1];
my_ull(&b[0], &op->db_id);
execute_wrap(stm_stream_rm_file, b, NULL);
}

3
recording-daemon/db.h
Unescape View File

@ -10,6 +10,9 @@ void db_do_stream(metafile_t *mf, output_t *op, stream_t *, unsigned long ssrc);
void db_close_stream(output_t *op);
void db_delete_stream(metafile_t *, output_t *op);
void db_config_stream(output_t *op);
void db_set_s3(unsigned long long, const char *uri);
void db_set_gcs(unsigned long long, const char *uri);
void db_rm_file(output_t *);
#endif

38
recording-daemon/gcs.c
Unescape View File

@ -1,9 +1,13 @@
#include "gcs.h"
#include <json-glib/json-glib.h>
#include "notify.h"
#include "main.h"
#include "output.h"
#include "http.h"
#include "oauth.h"
#include "db.h"
static oauth_context_t auth_ctx;
@ -27,6 +31,29 @@ static void gcs_cleanup(notif_req_t *req) {
}
static char *get_uri(const GString *response, const char *field) {
g_autoptr(JsonParser) parser = json_parser_new();
g_autoptr(GError) error = NULL;
json_parser_load_from_data(parser, response->str, response->len, &error);
if (error)
return NULL;
JsonNode *root = json_parser_get_root(parser);
if (!root)
return NULL;
JsonObject *o = json_node_get_object(root);
if (!o)
return NULL;
const char *s = json_object_get_string_member(o, field);
if (!s)
return NULL;
return g_strdup(s);
}
static bool gcs_perform(notif_req_t *req) {
if (!req->content) {
ilog(LOG_ERR, "Content for GCS upload unavailable ('%s%s%s')", FMT_M(req->name));
@ -56,6 +83,7 @@ static bool gcs_perform(notif_req_t *req) {
g_autoptr(GString) response = g_string_new("");
g_autoptr(char) uri;
g_autoptr(char) access_uri = NULL;
if (gcs_key && gcs_key[0])
uri = g_strdup_printf("%s?name=%s&uploadType=media&key=%s",
@ -97,6 +125,16 @@ static bool gcs_perform(notif_req_t *req) {
ilog(LOG_DEBUG, "GCS upload for '%s%s%s' successful", FMT_M(req->name));
// extract access URI
if (gcs_medialink)
access_uri = get_uri(response, "mediaLink");
if (!access_uri)
access_uri = get_uri(response, "selfLink");
if (!access_uri)
access_uri = g_strdup_printf("%s/%s", gcs_uri, req->object_name);
db_set_gcs(req->db_id, access_uri);
return true;
err:

2
recording-daemon/main.c
Unescape View File

@ -83,6 +83,7 @@ char *gcs_key;
char *gcs_service_account;
char *gcs_scope;
gboolean gcs_nverify;
gboolean gcs_medialink;
static GQueue threads = G_QUEUE_INIT; // only accessed from main thread
@ -268,6 +269,7 @@ static void options(int *argc, char ***argv) {
{ "gcs-service-account", 0, 0, G_OPTION_ARG_FILENAME, &gcs_service_account,"Service account JSON file for GCS JWT authentication","FILE" },
{ "gcs-scope", 0, 0, G_OPTION_ARG_STRING, &gcs_scope, "Scope for GCS JWT authentication", "STRING" },
{ "gcs-no-verify", 0, 0, G_OPTION_ARG_NONE, &gcs_nverify, "Disable TLS verification for GCS", NULL },
{ "gcs-media-link", 0, 0, G_OPTION_ARG_NONE, &gcs_medialink, "Store direct access URI", NULL },
{ NULL, }
};

1
recording-daemon/main.h
Unescape View File

@ -72,6 +72,7 @@ extern char *gcs_key;
extern char *gcs_service_account;
extern char *gcs_scope;
extern gboolean gcs_nverify;
extern gboolean gcs_medialink;
extern struct rtpengine_common_config rtpe_common_config;

2
recording-daemon/output.c
Unescape View File

@ -692,6 +692,8 @@ void output_close(metafile_t *mf, output_t *output, tag_t *tag, bool discard) {
if (unlink(output->filename))
ilog(LOG_ERR, "Failed to delete file '%s%s%s': %s",
FMT_M(output->filename), strerror(errno));
else
db_rm_file(output);
}
g_clear_pointer(&output->full_filename, g_free);

3
recording-daemon/s3.c
Unescape View File

@ -4,6 +4,7 @@
#include "notify.h"
#include "http.h"
#include "s3utils.h"
#include "db.h"
static void s3_setup(notif_req_t *req, output_t *o, metafile_t *mf, tag_t *tag) {
@ -120,6 +121,8 @@ static bool s3_perform(notif_req_t *req) {
curl_slist_free_all(headers);
db_set_s3(req->db_id, uri);
return true;
fail:

Write Preview
Loading…
Cancel
Save