PVE/Status/InfluxDB.pm

   1 package PVE::Status::InfluxDB;
   2
   3 use strict;
   4 use warnings;
   5
   6 use POSIX qw(isnan isinf);
   7 use Scalar::Util 'looks_like_number';
   8 use IO::Socket::IP;
   9 use LWP::UserAgent;
  10 use HTTP::Request;
  11
  12 use PVE::SafeSyslog;
  13
  14 use PVE::Status::Plugin;
  15
  16 use base('PVE::Status::Plugin');
  17
  18 sub type {
  19     return 'influxdb';
  20 }
  21
  22 sub properties {
  23     return {
  24         organization => {
  25             description => "The InfluxDB organization. Only necessary when using the http v2 api."
  26                ." Has no meaning when using v2 compatibility api.",
  27             type => 'string',
  28             optional => 1,
  29         },
  30         bucket => {
  31             description => "The InfluxDB bucket/db. Only necessary when using the http v2 api.",
  32             type => 'string',
  33             optional => 1,
  34         },
  35         token => {
  36             description => "The InfluxDB access token. Only necessary when using the http v2 api."
  37                 ." If the v2 compatibility api is used, use 'user:password' instead.",
  38             type => 'string',
  39             optional => 1,
  40         },
  41         'api-path-prefix' => {
  42             description => "An API path prefix inserted between '<host>:<port>/' and '/api2/'."
  43                 ." Can be useful if the InfluxDB service runs behind a reverse proxy.",
  44             type => 'string',
  45             optional => 1,
  46         },
  47         influxdbproto => {
  48             type => 'string',
  49             enum => ['udp', 'http', 'https'],
  50             default => 'udp',
  51             optional => 1,
  52         },
  53         'max-body-size' => {
  54             description => "InfluxDB max-body-size in bytes. Requests are batched up to this size.",
  55             type => 'integer',
  56             minimum => 1,
  57             default => 25_000_000,
  58         },
  59         'verify-certificate' => {
  60             description => "Set to 0 to disable certificate verification for https endpoints.",
  61             type => 'boolean',
  62             optional => 1,
  63             default => 1,
  64         },
  65     };
  66 }
  67 sub options {
  68     return {
  69         server => {},
  70         port => {},
  71         mtu => { optional => 1 },
  72         disable => { optional => 1 },
  73         organization => { optional => 1},
  74         bucket => { optional => 1},
  75         token => { optional => 1},
  76         influxdbproto => { optional => 1},
  77         timeout => { optional => 1},
  78         'max-body-size' => { optional => 1 },
  79         'api-path-prefix' => { optional => 1 },
  80         'verify-certificate' => { optional => 1 },
  81    };
  82 }
  83
  84 my $set_ssl_opts = sub {
  85     my ($cfg, $ua) = @_;
  86
  87     my $cert_verify = $cfg->{'verify-certificate'} // 1;
  88     if (!$cert_verify) {
  89         $ua->ssl_opts(
  90             verify_hostname => 0,
  91             SSL_verify_mode => IO::Socket::SSL::SSL_VERIFY_NONE,
  92         );
  93     }
  94
  95     return;
  96 };
  97
  98 # Plugin implementation
  99 sub update_node_status {
 100     my ($class, $txn, $node, $data, $ctime) = @_;
 101
 102     $ctime *= 1000000000;
 103
 104     build_influxdb_payload($class, $txn, $data, $ctime, "object=nodes,host=$node");
 105 }
 106
 107 sub update_qemu_status {
 108     my ($class, $txn, $vmid, $data, $ctime, $nodename) = @_;
 109
 110     $ctime *= 1000000000;
 111
 112     my $object = "object=qemu,vmid=$vmid,nodename=$nodename";
 113     if($data->{name} && $data->{name} ne '') {
 114         $object .= ",host=$data->{name}";
 115     }
 116     $object =~ s/\s/\\ /g;
 117
 118     # VMID is already added in base $object above, so exclude it from being re-added
 119     build_influxdb_payload($class, $txn, $data, $ctime, $object, { 'vmid' => 1 });
 120 }
 121
 122 sub update_lxc_status {
 123     my ($class, $txn, $vmid, $data, $ctime, $nodename) = @_;
 124
 125     $ctime *= 1000000000;
 126
 127     my $object = "object=lxc,vmid=$vmid,nodename=$nodename";
 128     if($data->{name} && $data->{name} ne '') {
 129         $object .= ",host=$data->{name}";
 130     }
 131     $object =~ s/\s/\\ /g;
 132
 133     # VMID is already added in base $object above, so exclude it from being re-added
 134     build_influxdb_payload($class, $txn, $data, $ctime, $object, { 'vmid' => 1 });
 135 }
 136
 137 sub update_storage_status {
 138     my ($class, $txn, $nodename, $storeid, $data, $ctime) = @_;
 139
 140     $ctime *= 1000000000;
 141
 142     my $object = "object=storages,nodename=$nodename,host=$storeid";
 143     if($data->{type} && $data->{type} ne '') {
 144         $object .= ",type=$data->{type}";
 145     }
 146     $object =~ s/\s/\\ /g;
 147
 148     build_influxdb_payload($class, $txn, $data, $ctime, $object);
 149 }
 150
 151 sub _send_batch_size {
 152     my ($class, $cfg) = @_;
 153     my $proto = $cfg->{influxdbproto} // 'udp';
 154     if ($proto ne 'udp') {
 155         return $cfg->{'max-body-size'} // 25_000_000;
 156     }
 157
 158     return $class->SUPER::_send_batch_size($cfg);
 159 }
 160
 161 sub send {
 162     my ($class, $connection, $data, $cfg) = @_;
 163
 164     my $proto = $cfg->{influxdbproto} // 'udp';
 165     if ($proto eq 'udp') {
 166         return $class->SUPER::send($connection, $data, $cfg);
 167     } elsif ($proto =~ m/^https?$/) {
 168         my $ua = LWP::UserAgent->new();
 169         $set_ssl_opts->($cfg, $ua);
 170         $ua->timeout($cfg->{timeout} // 1);
 171         $connection->content($data);
 172         my $response = $ua->request($connection);
 173
 174         if (!$response->is_success) {
 175             my $err = $response->status_line;
 176             die "$err\n";
 177         }
 178     } else {
 179         die "invalid protocol\n";
 180     }
 181
 182     return;
 183 }
 184
 185 sub _disconnect {
 186     my ($class, $connection, $cfg) = @_;
 187     my $proto = $cfg->{influxdbproto} // 'udp';
 188     if ($proto eq 'udp') {
 189         return $class->SUPER::_disconnect($connection, $cfg);
 190     }
 191
 192     return;
 193 }
 194
 195 sub _get_v2url {
 196     my ($cfg, $api_path) = @_;
 197     my ($proto, $host, $port) = $cfg->@{qw(influxdbproto server port)};
 198     my $api_prefix = $cfg->{'api-path-prefix'} // '/';
 199     if ($api_prefix ne '/' && $api_prefix =~ m!^/*(.+)/*$!) {
 200         $api_prefix = "/$1/";
 201     }
 202
 203     if ($api_path ne 'health') {
 204         $api_path = "api/v2/${api_path}";
 205     }
 206
 207     return "${proto}://${host}:${port}${api_prefix}${api_path}";
 208 }
 209
 210 sub _connect {
 211     my ($class, $cfg, $id) = @_;
 212
 213     my $host = $cfg->{server};
 214     my $port = $cfg->{port};
 215     my $proto = $cfg->{influxdbproto} // 'udp';
 216
 217     if ($proto eq 'udp') {
 218         my $socket = IO::Socket::IP->new(
 219             PeerAddr    => $host,
 220             PeerPort    => $port,
 221             Proto       => 'udp',
 222         ) || die "couldn't create influxdb socket [$host]:$port - $@\n";
 223
 224         $socket->blocking(0);
 225
 226         return $socket;
 227     } elsif ($proto =~ m/^https?$/) {
 228         my $token = get_credentials($id);
 229         my $org = $cfg->{organization} // 'proxmox';
 230         my $bucket = $cfg->{bucket} // 'proxmox';
 231         my $url = _get_v2url($cfg, "write?org=${org}&bucket=${bucket}");
 232
 233         my $req = HTTP::Request->new(POST => $url);
 234         if (defined($token)) {
 235             $req->header( "Authorization", "Token $token");
 236         }
 237
 238         return $req;
 239     }
 240
 241     die "cannot connect to InfluxDB: invalid protocol '$proto'\n";
 242 }
 243
 244 sub test_connection {
 245     my ($class, $cfg, $id) = @_;
 246
 247     # do not check connection for disabled plugins
 248     return if $cfg->{disable};
 249
 250     my $proto = $cfg->{influxdbproto} // 'udp';
 251     if ($proto eq 'udp') {
 252         return $class->SUPER::test_connection($cfg, $id);
 253     } elsif ($proto =~ m/^https?$/) {
 254         my $url = _get_v2url($cfg, "health");
 255         my $ua = LWP::UserAgent->new();
 256         $set_ssl_opts->($cfg, $ua);
 257         $ua->timeout($cfg->{timeout} // 1);
 258         # in the initial add connection test, the token may still be in $cfg
 259         my $token = $cfg->{token} // get_credentials($id, 1);
 260         if (defined($token)) {
 261             $ua->default_header("Authorization" => "Token $token");
 262         }
 263         my $response = $ua->get($url);
 264
 265         if (!$response->is_success) {
 266             my $err = $response->status_line;
 267             die "$err\n";
 268         }
 269     } else {
 270         die "invalid protocol\n";
 271     }
 272
 273     return;
 274 }
 275
 276 sub build_influxdb_payload {
 277     my ($class, $txn, $data, $ctime, $tags, $excluded, $measurement, $instance) = @_;
 278
 279     my @values = ();
 280
 281     foreach my $key (sort keys %$data) {
 282         next if defined($excluded) && $excluded->{$key};
 283         my $value = $data->{$key};
 284         next if !defined($value);
 285
 286         if (!ref($value) && $value ne '') {
 287             # value is scalar
 288
 289             if (defined(my $v = prepare_value($value))) {
 290                 push @values, "$key=$v";
 291             }
 292         } elsif (ref($value) eq 'HASH') {
 293             # value is a hash
 294
 295             if (!defined($measurement)) {
 296                 build_influxdb_payload($class, $txn, $value, $ctime, $tags, $excluded, $key);
 297             } elsif(!defined($instance)) {
 298                 build_influxdb_payload($class, $txn, $value, $ctime, $tags, $excluded, $measurement, $key);
 299             } else {
 300                 push @values, get_recursive_values($value);
 301             }
 302         }
 303     }
 304
 305     if (@values > 0) {
 306         my $mm = $measurement // 'system';
 307         my $tagstring = $tags;
 308         $tagstring .= ",instance=$instance" if defined($instance);
 309         my $valuestr = join(',', @values);
 310         $class->add_metric_data($txn, "$mm,$tagstring $valuestr $ctime\n");
 311     }
 312 }
 313
 314 sub get_recursive_values {
 315     my ($hash) = @_;
 316
 317     my @values = ();
 318
 319     foreach my $key (keys %$hash) {
 320         my $value = $hash->{$key};
 321         if(ref($value) eq 'HASH') {
 322             push(@values, get_recursive_values($value));
 323         } elsif (!ref($value) && $value ne '') {
 324             if (defined(my $v = prepare_value($value))) {
 325                 push @values, "$key=$v";
 326             }
 327         }
 328     }
 329
 330     return @values;
 331 }
 332
 333 sub prepare_value {
 334     my ($value) = @_;
 335
 336     if (looks_like_number($value)) {
 337         if (isnan($value) || isinf($value)) {
 338             # we cannot send influxdb NaN or Inf
 339             return undef;
 340         }
 341
 342         # influxdb also accepts 1.0e+10, etc.
 343         return $value;
 344     }
 345
 346     # non-numeric values require to be quoted, so escape " with \"
 347     $value =~ s/\"/\\\"/g;
 348     $value = "\"$value\"";
 349
 350     return $value;
 351 }
 352
 353 my $priv_dir = "/etc/pve/priv/metricserver";
 354
 355 sub cred_file_name {
 356     my ($id) = @_;
 357     return "${priv_dir}/${id}.pw";
 358 }
 359
 360 sub delete_credentials {
 361     my ($id) = @_;
 362
 363     if (my $cred_file = cred_file_name($id)) {
 364         unlink($cred_file)
 365             or warn "removing influxdb credentials file '$cred_file' failed: $!\n";
 366     }
 367
 368     return;
 369 }
 370
 371 sub set_credentials {
 372     my ($id, $token) = @_;
 373
 374     my $cred_file = cred_file_name($id);
 375
 376     mkdir $priv_dir;
 377
 378     PVE::Tools::file_set_contents($cred_file, "$token");
 379 }
 380
 381 sub get_credentials {
 382     my ($id, $silent) = @_;
 383
 384     my $cred_file = cred_file_name($id);
 385
 386     my $creds = eval { PVE::Tools::file_get_contents($cred_file) };
 387     warn "could not load credentials for '$id': $@\n" if $@ && !$silent;
 388
 389     return $creds;
 390 }
 391
 392 sub on_add_hook {
 393     my ($class, $id, $opts, $sensitive_opts) = @_;
 394
 395     my $token = $sensitive_opts->{token};
 396
 397     if (defined($token)) {
 398         set_credentials($id, $token);
 399     } else {
 400         delete_credentials($id);
 401     }
 402
 403     return undef;
 404 }
 405
 406 sub on_update_hook {
 407     my ($class, $id, $opts, $sensitive_opts) = @_;
 408     return if !exists($sensitive_opts->{token});
 409
 410     my $token = $sensitive_opts->{token};
 411     if (defined($token)) {
 412         set_credentials($id, $token);
 413     } else {
 414         delete_credentials($id);
 415     }
 416
 417     return undef;
 418 }
 419
 420 sub on_delete_hook {
 421     my ($class, $id, $opts) = @_;
 422
 423     delete_credentials($id);
 424
 425     return undef;
 426 }
 427
 428
 429 1;