src/PVE/Jobs/RealmSync.pm

   1 package PVE::Jobs::RealmSync;
   2
   3 use strict;
   4 use warnings;
   5
   6 use JSON qw(decode_json encode_json);
   7 use POSIX qw(ENOENT);
   8
   9 use PVE::JSONSchema qw(get_standard_option);
  10 use PVE::Cluster ();
  11 use PVE::CalendarEvent ();
  12 use PVE::Tools ();
  13
  14 use PVE::API2::Domains ();
  15
  16 # load user-* standard options
  17 use PVE::API2::User ();
  18
  19 use base qw(PVE::Job::Registry);
  20
  21 sub type {
  22     return 'realm-sync';
  23 }
  24
  25 my $props = get_standard_option('realm-sync-options', {
  26     realm => get_standard_option('realm'),
  27 });
  28
  29 sub properties {
  30     return $props;
  31 }
  32
  33 sub options {
  34     my $options = {
  35         enabled => { optional => 1 },
  36         schedule => {},
  37         comment => { optional => 1 },
  38         scope => {},
  39     };
  40     for my $opt (keys %$props) {
  41         next if defined($options->{$opt});
  42         # ignore legacy props from realm-sync schema
  43         next if $opt eq 'full' || $opt eq 'purge';
  44         if ($props->{$opt}->{optional}) {
  45             $options->{$opt} = { optional => 1 };
  46         } else {
  47             $options->{$opt} = {};
  48         }
  49     }
  50     $options->{realm}->{fixed} = 1;
  51
  52     return $options;
  53 }
  54
  55 sub decode_value {
  56     my ($class, $type, $key, $value) = @_;
  57     return $value;
  58 }
  59
  60 sub encode_value {
  61     my ($class, $type, $key, $value) = @_;
  62     return $value;
  63 }
  64
  65 sub createSchema {
  66     my ($class, $skip_type) = @_;
  67
  68     my $schema = $class->SUPER::createSchema($skip_type);
  69
  70     my $opts = $class->options();
  71     for my $opt (keys $schema->{properties}->%*) {
  72         next if defined($opts->{$opt}) || $opt eq 'id';
  73         delete $schema->{properties}->{$opt};
  74     }
  75
  76     return $schema;
  77 }
  78
  79 sub updateSchema {
  80     my ($class, $skip_type) = @_;
  81     my $schema = $class->SUPER::updateSchema($skip_type);
  82
  83     my $opts = $class->options();
  84     for my $opt (keys $schema->{properties}->%*) {
  85         next if defined($opts->{$opt});
  86         next if $opt eq 'id' || $opt eq 'delete';
  87         delete $schema->{properties}->{$opt};
  88     }
  89
  90     return $schema;
  91 }
  92
  93 my $statedir = "/etc/pve/priv/jobs";
  94
  95 sub get_state {
  96     my ($id) = @_;
  97
  98     mkdir $statedir;
  99     my $statefile = "$statedir/realm-sync-$id.json";
 100     my $raw = eval { PVE::Tools::file_get_contents($statefile) } // '';
 101
 102     my $state = ($raw =~ m/^(\{.*\})$/) ? decode_json($1) : {};
 103
 104     return $state;
 105 }
 106
 107 sub save_state {
 108     my ($id, $state) = @_;
 109
 110     mkdir $statedir;
 111     my $statefile = "$statedir/realm-sync-$id.json";
 112
 113     if (defined($state)) {
 114         PVE::Tools::file_set_contents($statefile, encode_json($state));
 115     } else {
 116         unlink $statefile or $! == ENOENT or die "could not delete state for $id - $!\n";
 117     }
 118
 119     return undef;
 120 }
 121
 122 sub run {
 123     my ($class, $conf, $id, $schedule) = @_;
 124
 125     for my $opt (keys %$conf) {
 126         delete $conf->{$opt} if !defined($props->{$opt});
 127     }
 128
 129     my $realm = $conf->{realm};
 130
 131     # cluster synced
 132     my $now = time();
 133     my $nodename = PVE::INotify::nodename();
 134
 135     # check statefile in pmxcfs if we should start
 136     my $shouldrun = PVE::Cluster::cfs_lock_domain('realm-sync', undef, sub {
 137         my $members = PVE::Cluster::get_members();
 138
 139         my $state = get_state($id);
 140         my $last_node = $state->{node} // $nodename;
 141         my $last_upid = $state->{upid};
 142         my $last_time = $state->{time};
 143
 144         my $last_node_online = $last_node eq $nodename || ($members->{$last_node} // {})->{online};
 145
 146         if (defined($last_upid)) {
 147             # first check if the next run is scheduled
 148             if (my $parsed = PVE::Tools::upid_decode($last_upid, 1)) {
 149                 my $cal_spec = PVE::CalendarEvent::parse_calendar_event($schedule);
 150                 my $next_sync = PVE::CalendarEvent::compute_next_event($cal_spec, $parsed->{starttime});
 151                 return 0 if !defined($next_sync) || $now < $next_sync; # not yet its (next) turn
 152             }
 153             # check if still running and node is online
 154             my $tasks = PVE::Cluster::get_tasklist();
 155             for my $task (@$tasks) {
 156                 next if $task->{upid} ne $last_upid;
 157                 last if defined($task->{endtime}); # it's already finished
 158                 last if !$last_node_online; # it's not finished and the node is offline
 159                 return 0; # not finished and online
 160             }
 161         } elsif (defined($last_time) && ($last_time+60) > $now && $last_node_online) {
 162             # another node started this job in the last 60 seconds and is still online
 163             return 0;
 164         }
 165
 166         # any of the following conditions should be true here:
 167         # * it was started on another node but that node is offline now
 168         # * it was started but either too long ago, or with an error
 169         # * the started task finished
 170
 171         save_state($id, {
 172             node => $nodename,
 173             time => $now,
 174         });
 175         return 1;
 176     });
 177     die $@ if $@;
 178
 179     if ($shouldrun) {
 180         my $upid = eval { PVE::API2::Domains->sync($conf) };
 181         my $err = $@;
 182         PVE::Cluster::cfs_lock_domain('realm-sync', undef, sub {
 183             if ($err && !$upid) {
 184                 save_state($id, {
 185                     node => $nodename,
 186                     time => $now,
 187                     error => $err,
 188                 });
 189                 die "$err\n";
 190             }
 191
 192             save_state($id, {
 193                 node => $nodename,
 194                 upid => $upid,
 195             });
 196         });
 197         die $@ if $@;
 198         return $upid;
 199     }
 200
 201     return "OK"; # all other cases should not run the sync on this node
 202 }
 203
 204 1;