package MogileFS::Worker::Monitor;
use strict;
use warnings;

use base 'MogileFS::Worker';
use fields (
            'last_test_write', # devid -> time.  time we last tried writing to a device.
            'monitor_start',   # main monitor start time
            'skip_host',       # hostid -> 1 if already noted dead (reset every loop)
            'seen_hosts',      # IP -> 1 (reset every loop)
            'iow',             # MogileFS::IOStatWatcher object
            'prev_data',       # DB data from previous run
            'devutil',         # Running tally of device utilization
            'events',          # Queue of state events
            'refresh_state',   # devid -> { used, total, callbacks }, temporary data in each refresh run
            'have_masterdb',   # Hint flag for if the master DB is available
            'updateable_devices', # devid -> Device, avoids device table updates
            'parent',          # socketpair to parent process
            'refresh_pending', # set if there was a manually-requested refresh
            'db_monitor_ran',  # We announce "monitor_just_ran" every time the
                               # device checks are run, but only if the DB has
                               # been checked inbetween.
            'devs_to_update'   # device table update queue
            );

use Danga::Socket 1.56;
use MogileFS::Config;
use MogileFS::Util qw(error debug encode_url_args apply_state_events_list);
use MogileFS::IOStatWatcher;
use MogileFS::Server;
use MogileFS::Connection::Parent;
use Digest::MD5 qw(md5_base64);

use constant UPDATE_DB_EVERY => 15;

sub new {
    my ($class, $psock) = @_;
    my $self = fields::new($class);
    $self->SUPER::new($psock);

    $self->{last_test_write} = {};
    $self->{iow}             = MogileFS::IOStatWatcher->new;
    $self->{prev_data}       = { domain => {}, class => {}, host => {},
        device => {} };
    $self->{devutil}         = { cur => {}, prev => {}, tmp => {} };
    $self->{events}          = [];
    $self->{have_masterdb}   = 0;
    return $self;
}

sub watchdog_timeout {
    30;
}

# returns 1 if a DB update was attempted
# returns 0 immediately if the (device) monitor is already running
sub cache_refresh {
    my $self = shift;

    if ($self->{refresh_state}) {
        debug("Monitor run in progress, will not check for DB updates");
        return 0;
    }

    debug("Monitor running; checking DB for updates");
    # "Fix" our local cache of this flag, so we always check the master DB.
    MogileFS::Config->cache_server_setting('_master_db_alive', 1);
    my $have_dbh = $self->validate_dbh;
    if ($have_dbh && !$self->{have_masterdb}) {
        $self->{have_masterdb} = 1;
        $self->set_event('srvset', '_master_db_alive', { value => 1 });
    } elsif (!$have_dbh) {
        $self->{have_masterdb} = 0;
        $self->set_event('srvset', '_master_db_alive', { value => 0 });
        error("Cannot connect to master database!");
    }

    if ($have_dbh) {
        my $db_data   = $self->grab_all_data;

        # Stack diffs to ship back later
        $self->diff_data($db_data);
    }

    $self->send_events_to_parent;
    $self->{db_monitor_ran} = 1;

    return 1;
}

sub usage_refresh {
    my ($self) = @_;

    # prevent concurrent refresh
    return if $self->{refresh_state};

    debug("Monitor running; scanning usage files");

    $self->{refresh_state} = {}; # devid -> ...
    $self->{monitor_start} = Time::HiRes::time();

    my $have_dbh = $self->validate_dbh;

    # See if we should be allowed to update the device table rows.
    if ($have_dbh && Mgd::get_store()->get_lock('mgfs:device_update', 0)) {
        # Fetch the freshlist list of entries, to avoid excessive writes.
        $self->{updateable_devices} = { map { $_->{devid} => $_ }
            Mgd::get_store()->get_all_devices };
        $self->{devs_to_update} = [];
    } else {
        $self->{updateable_devices} = undef;
    }

    $self->{skip_host}  = {};  # hostid -> 1 if already noted dead.
    $self->{seen_hosts} = {}; # IP -> 1

    my $dev_factory = MogileFS::Factory::Device->get_factory();
    my $devutil = $self->{devutil};

    $devutil->{tmp} = {};
    # kick off check_device to test host/devs. diff against old values.
    for my $dev ($dev_factory->get_all) {
        if (my $state = $self->is_iow_diff($dev)) {
            $self->state_event('device', $dev->id, {utilization => $state});
        }
        $devutil->{tmp}->{$dev->id} = $devutil->{cur}->{$dev->id};

        $dev->can_read_from or next;
        $self->check_device_begin($dev);
    }
    # we're done if we didn't schedule any work
    $self->usage_refresh_done unless keys %{$self->{refresh_state}};
}

sub usage_refresh_done {
    my ($self) = @_;

    $self->{devutil}->{prev} = $self->{devutil}->{tmp};
    # Set the IOWatcher hosts (once old monitor code has been disabled)

    $self->send_events_to_parent;

    $self->{iow}->set_hosts(keys %{$self->{seen_hosts}});

    foreach my $devid (keys %{$self->{refresh_state}}) {
        error("device check incomplete for dev$devid");
    }

    my $start = delete $self->{monitor_start};
    my $elapsed = Time::HiRes::time() - $start;
    debug("device refresh finished after $elapsed");

    $self->{refresh_state} = undef;
    my $pending_since = $self->{refresh_pending};

    # schedule another usage_refresh immediately if somebody requested it
    # Don't announce :monitor_just_ran if somebody requested a refresh
    # while we were running, we could've been refreshing on a stale DB
    if ($pending_since && $pending_since > $start) {
        # using AddTimer to schedule the refresh to avoid stack overflow
        # since usage_refresh can call usage_refresh_done directly if
        # there are no devices
        Danga::Socket->AddTimer(0, sub {
            $self->cache_refresh;
            $self->usage_refresh;
        });
    }

    # announce we're done if we ran on schedule, or we had a
    # forced refresh that was requested before we started.
    if (!$pending_since || $pending_since <= $start) {
        # totally done refreshing, accept manual refresh requests again
        $self->{parent}->watch_read(1);
        delete $self->{refresh_pending};
        if (delete $self->{db_monitor_ran} || $pending_since) {
            $self->send_to_parent(":monitor_just_ran");
        }
    }

    if ($self->{updateable_devices}) {
        my $sto = Mgd::get_store();
        my $updates = delete $self->{devs_to_update};
        $sto->update_device_usages($updates, sub { $self->still_alive });
        $sto->release_lock('mgfs:device_update');
        $self->{updateable_devices} = undef;
    }
}

sub work {
    my $self = shift;

    # It makes sense to have monitor use a shorter timeout
    # (conn_timeout) across the board to skip slow hosts.  Other workers
    # are less tolerant, and may use a higher value in node_timeout.
    MogileFS::Config->set_config_no_broadcast("node_timeout", MogileFS::Config->config("conn_timeout"));

    my $iow = $self->{iow};
    $iow->on_stats(sub {
        my ($hostname, $stats) = @_;

        while (my ($devid, $util) = each %$stats) {
            # Lets not propagate devices that we accidentally find.
            my $dev = Mgd::device_factory()->get_by_id($devid);
            next unless $dev;
            $self->{devutil}->{cur}->{$devid} = $util;
        }
    });

    my $db_monitor;
    $db_monitor = sub {
        $self->still_alive;

        # reschedule immediately if we were blocked by main_monitor.
        # setting refresh_pending will call cache_refresh again
        if (!$self->cache_refresh) {
            $self->{refresh_pending} ||= Time::HiRes::time();
        }

        # always reschedule in 4 seconds, regardless
        Danga::Socket->AddTimer(4, $db_monitor);
    };

    $db_monitor->();
    $self->read_from_parent;

    my $main_monitor;
    $main_monitor = sub {
        $self->{parent}->ping;
        $self->usage_refresh;
        Danga::Socket->AddTimer(2.5, $main_monitor);
    };

    $self->parent_ping; # ensure we get the initial DB state back
    $self->{parent} = MogileFS::Connection::Parent->new($self);
    Danga::Socket->AddTimer(0, $main_monitor);
    Danga::Socket->EventLoop;
}

sub process_line {
    my MogileFS::Worker::Monitor $self = shift;
    my $lineref = shift;
    if ($$lineref =~ /^:refresh_monitor$/) {
        if ($self->cache_refresh) {
            $self->usage_refresh;
        } else {
            $self->{refresh_pending} ||= Time::HiRes::time();
        }
        # try to stop processing further refresh_monitor requests
        # if we're acting on a manual refresh
        $self->{parent}->watch_read(0);
        return 1;
    }
    return 0;
}

# --------------------------------------------------------------------------

# Flattens and flips events up to the parent. Can be huge on startup!
# Events: set type foo=bar&baz=quux
# remove type id
# setstate type id foo=bar&baz=quux
# Combined: ev_mode=set&ev_type=device&foo=bar
# ev_mode=setstate&ev_type=device&ev_id=1&foo=bar
sub send_events_to_parent {
    my $self = shift;
    my @flat = ();
    for my $ev (@{$self->{events}}) {
        my ($mode, $type, $args) = @$ev;
        $args->{ev_mode} = $mode;
        $args->{ev_type} = $type;
        push(@flat, encode_url_args($args));
    }
    return unless @flat;
    $self->{events} = [];

    {
        # $events can be several MB, so let it go out-of-scope soon:
        my $events = join(' ', ':monitor_events', @flat);
        debug("sending state changes $events", 2);
        $self->send_to_parent($events);
    }

    apply_state_events_list(@flat);
}

sub add_event {
    push(@{$_[0]->{events}}, $_[1]);
}

sub set_event { 
    # Allow callers to use shorthand
    $_[3]->{ev_id} = $_[2];
    $_[0]->add_event(['set', $_[1], $_[3]]); 
}
sub remove_event { $_[0]->add_event(['remove', $_[1], { ev_id => $_[2] }]); }
sub state_event {
    $_[3]->{ev_id} = $_[2];
    $_[0]->add_event(['setstate', $_[1], $_[3]]);
}

sub is_iow_diff {
    my ($self, $dev) = @_;
    my $devid = $dev->id;
    my $p = $self->{devutil}->{prev}->{$devid};
    my $c = $self->{devutil}->{cur}->{$devid};
    if ( ! defined $p || $p ne $c ) {
        return $c;
    }
    return undef;
}

sub diff_data {
    my ($self, $db_data) = @_;

    my $new_data  = {};
    my $prev_data = $self->{prev_data};
    for my $type (keys %{$db_data}) {
        my $d_data = $db_data->{$type};
        my $p_data = $prev_data->{$type};
        my $n_data = {};

        for my $item (@{$d_data}) {
            my $id = $type eq 'domain' ? $item->{dmid}
                : $type eq 'class'     ? $item->{dmid} . '-' . $item->{classid}
                : $type eq 'host'      ? $item->{hostid}
                : $type eq 'device'    ? $item->{devid}
                : $type eq 'srvset'    ? $item->{field}
                : die "Unknown type";
            my $old = delete $p_data->{$id};
            # Special case: for devices, we don't care if mb_asof changes.
            # FIXME: Change the grab routine (or filter there?).
            delete $item->{mb_asof} if $type eq 'device';
            if (!$old || $self->diff_hash($old, $item)) {
                $self->set_event($type, $id, { %$item });
            }
            $n_data->{$id} = $item;
        }
        for my $id (keys %{$p_data}) {
            $self->remove_event($type, $id);
        }

        $new_data->{$type} = $n_data;
    }
    $self->{prev_data} = $new_data;
}

# returns 1 if the hashes are different.
sub diff_hash {
    my ($self, $old, $new) = @_;

    my %keys = ();
    map { $keys{$_}++ } keys %$old, keys %$new;
    for my $k (keys %keys) {
        return 1 if (exists $old->{$k} && ! exists $new->{$k});
        return 1 if (exists $new->{$k} && ! exists $old->{$k});
        return 1 if (defined $old->{$k} && ! defined $new->{$k});
        return 1 if (defined $new->{$k} && ! defined $old->{$k});
        next     if (! defined $new->{$k} && ! defined $old->{$k});
        return 1 if ($old->{$k} ne $new->{$k});
    }
    return 0;
}

sub grab_all_data {
    my $self = shift;
    my $sto  = Mgd::get_store();

    # Normalize the domain data to the rest to simplify the differ.
    # FIXME: Once new objects are swapped in, fix the original
    my %dom = $sto->get_all_domains;
    my @fixed_dom = ();
    while (my ($name, $id) = each %dom) {
        push(@fixed_dom, { namespace => $name, dmid => $id });
    }

    my $set = $sto->server_settings;
    my @fixed_set = ();
    while (my ($field, $value) = each %$set) {
        push(@fixed_set, { field => $field, value => $value });
    }

    my %ret = ( domain => \@fixed_dom,
        class  => [$sto->get_all_classes],
        host   => [$sto->get_all_hosts],
        device => [$sto->get_all_devices],
        srvset => \@fixed_set, );
    return \%ret;
}

# returns true on success, false on failure
sub check_usage_response {
    my ($self, $dev, $response) = @_;
    my $devid = $dev->id;

    my %stats;
    my $data = $response->content;
    foreach (split(/\r?\n/, $data)) {
        next unless /^(\w+)\s*:\s*(.+)$/;
        $stats{$1} = $2;
    }

    my ($used, $total) = ($stats{used}, $stats{total});
    unless ($used && $total) {
        $used  = "<undef>" unless defined $used;
        $total = "<undef>" unless defined $total;
        my $clen = length($data || "");
        error("dev$devid reports used = $used, total = $total, content-length: $clen, error?");
        return 0;
    }

    my $rstate = $self->{refresh_state}->{$devid};
    ($rstate->{used}, $rstate->{total}) = ($used, $total);

    # only update database every ~15 seconds per device
    if ($self->{updateable_devices}) {
        my $devrow = $self->{updateable_devices}->{$devid};
        my $last = ($devrow && $devrow->{mb_asof}) ? $devrow->{mb_asof} : 0;
        my $now = time();
        if ($last + UPDATE_DB_EVERY < $now) {
            my %upd = (mb_total => int($total / 1024),
                       mb_used  => int($used / 1024),
                       mb_asof => $now,
                       devid => $devid);
            push @{$self->{devs_to_update}}, \%upd;
        }
    }
    return 1;
}

sub dev_debug {
    my ($self, $dev, $writable) = @_;
    return unless $Mgd::DEBUG >= 1;
    my $devid = $dev->id;
    my $rstate = $self->{refresh_state}->{$devid};
    my ($used, $total) = ($rstate->{used}, $rstate->{total});

    debug("dev$devid: used = $used, total = $total, writeable = $writable");
}

sub check_write {
    my ($self, $dev) = @_;
    my $rstate = $self->{refresh_state}->{$dev->id};
    my $test_write = $rstate->{test_write};

    if (!$test_write || $test_write->{tries} > 0) {
        # this was "$$-$now" before, but we don't yet have a cleaner in
        # mogstored for these files
        my $num = int(rand 100);
        $test_write = $rstate->{test_write} ||= {};
        $test_write->{path} = "/dev${\$dev->id}/test-write/test-write-$num";
        $test_write->{content} = "time=" . time . " rand=$num";
        $test_write->{tries} ||= 2;
    }
    $test_write->{tries}--;

    my $opts = { content => $test_write->{content} };
    $dev->host->http("PUT", $test_write->{path}, $opts, sub {
        my ($response) = @_;
        $self->on_check_write_response($dev, $response);
    });
}

# starts the lengthy device check process
sub check_device_begin {
    my ($self, $dev) = @_;
    $self->{refresh_state}->{$dev->id} = {};

    $self->check_device($dev);
}

# the lengthy device check process
sub check_device {
    my ($self, $dev) = @_;
    return $self->check_device_done($dev) if $self->{skip_host}{$dev->hostid};

    my $devid = $dev->id;
    my $url = $dev->usage_url;
    my $host = $dev->host;

    $self->{seen_hosts}{$host->ip} = 1;

    # now try to get the data with a short timeout
    my $start_time = Time::HiRes::time();
    $host->http_get("GET", $dev->usage_url, undef, sub {
        my ($response) = @_;
        if (!$self->on_usage_response($dev, $response, $start_time)) {
            return $self->check_device_done($dev);
        }
        # next if we're not going to try this now
        my $now = time();
        if (($self->{last_test_write}{$devid} || 0) + UPDATE_DB_EVERY > $now) {
            return $self->check_device_done($dev);
        }
        $self->{last_test_write}{$devid} = $now;

        unless ($dev->can_delete_from) {
            # we should not try to write on readonly devices because it can be
            # mounted as RO.
            return $self->dev_observed_readonly($dev);
        }
        # now we want to check if this device is writeable

        # first, create the test-write directory.  this will return
        # immediately after the first time, as the 'create_directory'
        # function caches what it's already created.
        $dev->create_directory("/dev$devid/test-write", sub {
            $self->check_write($dev);
        });
    });
}

# called on a successful PUT, ensure the data we get back is what we uploaded
sub check_reread {
    my ($self, $dev) = @_;
    # now let's get it back to verify; note we use the get_port to
    # verify that the distinction works (if we have one)
    my $test_write = $self->{refresh_state}->{$dev->id}->{test_write};
    $dev->host->http_get("GET", $test_write->{path}, undef, sub {
        my ($response) = @_;
        $self->on_check_reread_response($dev, $response);
    });
}

sub on_check_reread_response {
    my ($self, $dev, $response) = @_;
    my $test_write = $self->{refresh_state}->{$dev->id}->{test_write};

    # if success and the content matches, mark it writeable
    if ($response->is_success) {
        if ($response->content eq $test_write->{content}) {
            if (!$dev->observed_writeable) {
                my $event = { observed_state => 'writeable' };
                $self->state_event('device', $dev->id, $event);
            }
            $self->dev_debug($dev, 1);

            return $self->check_bogus_md5($dev); # onto the final check...
        }

        # content didn't match due to race, retry and hope we're lucky
        return $self->check_write($dev) if ($test_write->{tries} > 0);
    }

    return $self->dev_observed_readonly($dev); # it's read-only at least
}

sub on_check_write_response {
    my ($self, $dev, $response) = @_;
    return $self->check_reread($dev) if $response->is_success;
    return $self->dev_observed_readonly($dev);
}

# returns true on success, false on failure
sub on_usage_response {
    my ($self, $dev, $response, $start_time) = @_;
    my $host = $dev->host;
    my $hostip = $host->ip;

    if ($response->is_success) {
        # at this point we can reach the host
        if (!$host->observed_reachable) {
            my $event = { observed_state => 'reachable' };
            $self->state_event('host', $dev->hostid, $event);
        }
        $self->{iow}->restart_monitoring_if_needed($hostip);

        return $self->check_usage_response($dev, $response);
    }

    my $url = $dev->usage_url;
    my $failed_after = Time::HiRes::time() - $start_time;
    if ($failed_after < 0.5) {
        if (!$dev->observed_unreachable) {
            my $event = { observed_state => 'unreachable' };
            $self->state_event('device', $dev->id, $event);
        }
        my $get_port = $host->http_get_port;
        error("Port $get_port not listening on $hostip ($url)?  Error was: " . $response->status_line);
    } else {
        $failed_after = sprintf("%.02f", $failed_after);
        if (!$host->observed_unreachable) {
            my $event = { observed_state => 'unreachable' };
            $self->state_event('host', $dev->hostid, $event);
        }
        $self->{skip_host}{$dev->hostid} = 1;
    }
    return 0; # failure
}

sub check_bogus_md5 {
    my ($self, $dev) = @_;
    my $put_path = "/dev${\$dev->id}/test-write/test-md5";
    my $opts = {
        headers => { "Content-MD5" => md5_base64("!") . "==", },
        content => '.',
    };

    # success is bad here, it means the server doesn't understand how to
    # verify and reject corrupt bodies from Content-MD5 headers.
    # most servers /will/ succeed here :<
    $dev->host->http("PUT", $put_path, $opts, sub {
        my ($response) = @_;
        $self->on_bogus_md5_response($dev, $response);
    });
}

sub on_bogus_md5_response {
    my ($self, $dev, $response) = @_;
    my $rej = $response->is_success ? 0 : 1;
    my $prev = $dev->reject_bad_md5;

    if (!defined($prev) || $prev != $rej) {
        debug("dev${\$dev->id}: reject_bad_md5 = $rej");
        $self->state_event('device', $dev->id, { reject_bad_md5 => $rej });
    }
    return $self->check_device_done($dev);
}

# if we fall through to here, then we know that something is not so
# good, so mark it readable which is guaranteed given we even tested
# writeability
sub dev_observed_readonly {
    my ($self, $dev) = @_;

    if (!$dev->observed_readable) {
        my $event = { observed_state => 'readable' };
        $self->state_event('device', $dev->id, $event);
    }
    $self->dev_debug($dev, 0);
    return $self->check_device_done($dev);
}

# called when all checks are done for a particular device
sub check_device_done {
    my ($self, $dev) = @_;

    $self->still_alive; # Ping parent if needed so we don't time out
                        # given lots of devices.
    delete $self->{refresh_state}->{$dev->id};

    # if refresh_state is totally empty, we're done
    if ((scalar keys %{$self->{refresh_state}}) == 0) {
        $self->usage_refresh_done;
    }
}

1;

# Local Variables:
# mode: perl
# c-basic-indent: 4
# indent-tabs-mode: nil
# End: