anchor.diff:                           X -> X
                        end,
       Trigger = trigger:init(trigger_periodic, fun () -> UpdateInterval end, ganglia_trigger),
  -    {_LastActive = 0, trigger:next(Trigger)}.
  +    {_LastActive = 0, trigger:next(Trigger), {0, 0, 0}}.
   
   -spec on(message(), state()) -> state().
  -on({ganglia_trigger}, {LastActive, TriggerOld}) ->
  -    Trigger = trigger:next(TriggerOld),
  -    gen_component:post_op(_State = {LastActive, Trigger}, _Message = {ganglia_periodic});
  +on({ganglia_trigger}, State) ->
  +    Trigger = trigger:next(get_trigger(State)),
  +    StateNew = inc_agg_id(set_trigger(Trigger, State)),
  +    gen_component:post_op(StateNew, _Msg = {ganglia_periodic});
   
   on({ganglia_periodic}, State) ->
  +    NewState = send_dht_node_metrics(State),
       send_general_metrics(),
  -    send_dht_node_metrics(),
       send_message_metrics(),
       send_rrd_metrics(),
       send_vivaldi_errors(),
  -    set_last_active(State).
  +    set_last_active(NewState);
  +
  +on({ganglia_dht_load_aggregation, AggId, Msg} ,State) ->
  +    ?TRACE("~p: ~p~n", [AggId, Msg]),
  +    {get_node_details_response, [{load, Load}]} = Msg,
  +    CurAggId = get_agg_id(State),
  +    if
  +        % check for correct aggregation id
  +        AggId == CurAggId  ->
  +            NewState = set_agg_load(Load, State),
  +            agg_check_pending(NewState);
  +        % otherwise drop this message
  +        true -> State
  +    end;
  +
  +on({ganglia_vivaldi_response, DHTName, Msg}, State) ->
  +    {vivaldi_get_coordinate_response, _, Confidence} = Msg,
  +    gmetric(both, lists:flatten(io_lib:format("vivaldi_confidence_~s", [DHTName])), "float", Confidence, "Confidence"),
  +    State.
   
   send_general_metrics() ->
       % general erlang status information
  
positive.code: 
  % @doc aggregate the number of key-value pairs
  %      and the amount of memory for this VM
  -spec send_dht_node_metrics() -> ok.
  send_dht_node_metrics() ->
      DHTNodes = pid_groups:find_all(dht_node),
      % Load of DHT Nodes
      Pairs = lists:foldl(fun (Pid, Agg) ->
                                  Agg + get_load(Pid)
                          end, 0, DHTNodes),
      _ = gmetric(both, "kv pairs", "int32", Pairs, "pairs"),
      % Memory Usage of DHT Nodes
      MemoryUsage = lists:sum([element(2, erlang:process_info(P, memory))
                               || P <- DHTNodes]),
      _ = gmetric(both, "Memory used by dht_nodes", "int32", MemoryUsage, "Bytes"),
      ok.
  
  -spec send_message_metrics() -> ok.
  send_message_metrics() ->
  
negative.code: 
  -export([start_link/1, init/1, on/2]).
  
  %-define(TRACE(X,Y), ok).
  -define(TRACE(X,Y), io:format(X, Y)).
  
  -type load_aggregation() :: {AggId :: non_neg_integer(), Pending :: non_neg_integer(), Load :: non_neg_integer()}.
  
  -type state() :: {LastUpdated :: non_neg_integer(), Trigger :: trigger:state(), LoadAggregation :: load_aggregation()}.
  -type message() :: {ganglia_trigger} | {ganglia_periodic} |
                     {ganglia_dht_node_aggregation, DHTNode :: pid(), message()} |
                     {vivaldi_get_coordinate_response, pid(), message()}.
  
  -spec start_link(pid_groups:groupname()) -> {ok, pid()}.
  start_link(ServiceGroup) ->
  