source: branches/HiveStatistics/sources/HeuristicLab.Services.Hive/3.3/HiveStatisticsGenerator.cs @ 12469

#region License Information
/* HeuristicLab
 * Copyright (C) 2002-2015 Heuristic and Evolutionary Algorithms Laboratory (HEAL)
 *
 * This file is part of HeuristicLab.
 *
 * HeuristicLab is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * HeuristicLab is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with HeuristicLab. If not, see <http://www.gnu.org/licenses/>.
 */
#endregion

using System;
using System.Collections.Generic;
using System.Data.Linq;
using System.Linq;
using HeuristicLab.Services.Hive.DataAccess;
using HeuristicLab.Services.Hive.DataAccess.Manager;

namespace HeuristicLab.Services.Hive {
  public class HiveStatisticsGenerator : IStatisticsGenerator {

    private const string UnknownUserName = "Unknown";
    private static readonly TimeSpan SmallestTimeSpan = new TimeSpan(0, 5, 0);

    public void GenerateStatistics() {
      using (var pm = new PersistenceManager()) {
        pm.UseTransaction(() => {
          UpdateDimUserTable(pm);
          UpdateDimJobTable(pm);
          UpdateDimClients(pm);
          pm.SubmitChanges();
        });

        pm.UseTransaction(() => {
          var newTime = UpdateDimTimeTable(pm);
          if (newTime != null) {
            pm.SubmitChanges();
            UpdateFactClientInfoTable(newTime, pm);
            UpdateTaskFacts(newTime, pm);
            try {
              pm.SubmitChanges();
            }
            catch (DuplicateKeyException e) {
              var logger = LogFactory.GetLogger(typeof(HiveStatisticsGenerator).Namespace);
              logger.Log(string.Format(
                @"Propable change from summertime to wintertime, resulting in overlapping times.
                          On wintertime to summertime change, slave timeouts and a fact gap will occur. 
                          Exception Details: {0}", e));
            }
          }
        });
      }
    }

    private DimTime UpdateDimTimeTable(PersistenceManager pm) {
      var dimTimeDao = pm.DimTimeDao;
      var now = DateTime.Now;
      var timeEntry = new DimTime {
        Time = new DateTime(now.Year, now.Month, now.Day, now.Hour, now.Minute, 0),
        Hour = new DateTime(now.Year, now.Month, now.Day, now.Hour, 0, 0),
        Day = new DateTime(now.Year, now.Month, now.Day, 0, 0, 0),
        Month = new DateTime(now.Year, now.Month, 1, 0, 0, 0),
        Year = new DateTime(now.Year, 1, 1, 0, 0, 0)
      };
      // using SaveOrAttach instead of save because the table can only be updated once every minute
      // (pk restriction) save would cause a duplicate key exception if the method is accidentally 
      // called more than once every minute (e.g. restarting the service within a minute)
      return dimTimeDao.SaveOrAttach(timeEntry);
    }

    private void UpdateDimUserTable(PersistenceManager pm) {
      var dimUserDao = pm.DimUserDao;
      var resourceDao = pm.ResourceDao;
      var jobDao = pm.JobDao;
      var existingUserIds = dimUserDao.GetAll().Select(x => x.UserId);
      var vaildResourceOwnerIds = resourceDao.GetResourcesWithValidOwner().Select(x => x.OwnerUserId.Value);
      var jobOwnerIds = jobDao.GetAll().Select(x => x.OwnerUserId);
      var newUserIds = vaildResourceOwnerIds
        .Union(jobOwnerIds)
        .Where(id => !existingUserIds.Contains(id))
        .ToList();
      dimUserDao.Save(newUserIds.Select(x => new DimUser {
        UserId = x,
        Name = GetUserName(x)
      }));
    }

    private void UpdateDimJobTable(PersistenceManager pm) {
      var dimJobDao = pm.DimJobDao;
      var jobDao = pm.JobDao;
      var dimJobIds = dimJobDao.GetAll().Select(x => x.JobId);
      var newJobs = jobDao.GetAll()
        .Where(x => !dimJobIds.Contains(x.JobId))
        .Select(x => new {
          JobId = x.JobId,
          UserId = x.OwnerUserId,
          JobName = x.Name ?? string.Empty,
        })
        .ToList();
      dimJobDao.Save(newJobs.Select(x => new DimJob {
        JobId = x.JobId,
        JobName = x.JobName,
        UserId = x.UserId,
        UserName = GetUserName(x.UserId)
      }));
    }

    private void UpdateDimClients(PersistenceManager pm) {
      var dimClientDao = pm.DimClientDao;
      var slaveDao = pm.SlaveDao;
      var slaves = slaveDao.GetAll();
      var recentlyAddedClients = dimClientDao.GetRecentlyAddedClients();
      var slaveIds = slaves.Select(x => x.ResourceId);

      var removedClientIds = recentlyAddedClients
        .Where(x => !slaveIds.Contains(x.ResourceId))
        .Select(x => x.Id);
      var modifiedClients =
        from slave in slaves
        join client in recentlyAddedClients on slave.ResourceId equals client.ResourceId
        where (slave.Name != client.Name
               || slave.ParentResourceId != client.ResourceGroupId
               || slave.ParentResource.ParentResourceId != client.ResourceGroup2Id)
        select new {
          SlaveId = slave.ResourceId,
          ClientId = client.Id
        };
      var clientIds = dimClientDao.GetAll().Select(x => x.ResourceId);
      var modifiedClientIds = modifiedClients.Select(x => x.SlaveId);
      var newClients = slaves
        .Where(x => !clientIds.Contains(x.ResourceId)
                    || modifiedClientIds.Contains(x.ResourceId))
        .Select(x => new {
          x.ResourceId,
          x.Name,
          ResourceGroupId = x.ParentResourceId,
          ResourceGroup2Id = x.ParentResource.ParentResourceId
        })
        .ToList();

      var clientsToUpdate = removedClientIds.Union(modifiedClients.Select(x => x.ClientId));
      dimClientDao.UpdateExpirationTime(clientsToUpdate, DateTime.Now);
      dimClientDao.Save(newClients.Select(x => new DimClient {
        ResourceId = x.ResourceId,
        Name = x.Name,
        ExpirationTime = null,
        ResourceGroupId = x.ResourceGroupId,
        ResourceGroup2Id = x.ResourceGroup2Id
      }));
    }

    private void UpdateFactClientInfoTable(DimTime newTime, PersistenceManager pm) {
      var factClientInfoDao = pm.FactClientInfoDao;
      var slaveDao = pm.SlaveDao;
      var dimClientDao = pm.DimClientDao;

      var newRawFactInfos =
        from s in slaveDao.GetAll()
        join c in dimClientDao.GetRecentlyAddedClients() on s.ResourceId equals c.ResourceId
        join lcf in factClientInfoDao.GetLastUpdateTimestamps() on c.ResourceId equals lcf.ResourceId into joinCf
        from cf in joinCf.DefaultIfEmpty()
        select new {
          ClientId = c.Id,
          UserId = s.OwnerUserId ?? Guid.Empty,
          TotalCores = s.Cores ?? 0,
          FreeCores = s.FreeCores ?? 0,
          TotalMemory = s.Memory ?? 0,
          FreeMemory = s.FreeMemory ?? 0,
          CpuUtilization = s.CpuUtilization,
          SlaveState = s.SlaveState,
          IsAllowedToCalculate = s.IsAllowedToCalculate,
          LastFactTimestamp = cf.Timestamp
        };

      factClientInfoDao.Save(
        from x in newRawFactInfos.ToList()
        let duration = x.LastFactTimestamp != null
                       ? (newTime.Time - (DateTime)x.LastFactTimestamp).TotalMinutes
                       : SmallestTimeSpan.TotalMinutes
        select new FactClientInfo {
          ClientId = x.ClientId,
          DimTime = newTime,
          UserId = x.UserId,
          NumUsedCores = x.TotalCores - x.FreeCores,
          NumTotalCores = x.TotalMemory,
          UsedMemory = x.TotalMemory - x.FreeMemory,
          TotalMemory = x.TotalMemory,
          CpuUtilization = x.CpuUtilization,
          SlaveState = x.SlaveState,
          TotalTimeIdle = x.SlaveState == SlaveState.Idle && x.IsAllowedToCalculate ? duration : 0.0,
          TotalTimeCalculating = x.SlaveState == SlaveState.Calculating ? duration : 0.0,
          TotalTimeUnavailable = x.SlaveState == SlaveState.Idle && !x.IsAllowedToCalculate ? duration : 0.0,
          TotalTimeOffline = x.SlaveState == SlaveState.Offline ? duration : 0.0
        }
      );
    }

    private void UpdateTaskFacts(DimTime newTime, PersistenceManager pm) {
      var factTaskDao = pm.FactTaskDao;
      var taskDao = pm.TaskDao;
      var dimClientDao = pm.DimClientDao;
      var stateLogDao = pm.StateLogDao;

      var factTaskIds = factTaskDao.GetAll().Select(x => x.TaskId);
      var notFinishedFactTasks = factTaskDao.GetNotFinishedTasks().Select(x => new {
        x.TaskId,
        x.DimTimeStart,
        x.LastClientId
      });

      var newTasks =
        from task in taskDao.GetAllChildTasks()
        let stateLogs = task.StateLogs.OrderByDescending(x => x.DateTime)
        let lastSlaveId = stateLogs.First(x => x.SlaveId != null).SlaveId
        where (!factTaskIds.Contains(task.TaskId)
               || notFinishedFactTasks.Select(x => x.TaskId).Contains(task.TaskId))
        join lastFactTask in notFinishedFactTasks on task.TaskId equals lastFactTask.TaskId into lastFactPerTask
        from lastFact in lastFactPerTask.DefaultIfEmpty()
        join client in dimClientDao.GetRecentlyAddedClients() on lastSlaveId equals client.ResourceId into clientsPerSlaveId
        from client in clientsPerSlaveId.DefaultIfEmpty()
        select new {
          TaskId = task.TaskId,
          JobId = task.JobId,
          Priority = task.Priority,
          CoresRequired = task.CoresNeeded,
          MemoryRequired = task.MemoryNeeded,
          State = task.State,
          StateLogs = stateLogs,
          DimTimeStart = lastFact != null ? lastFact.DimTimeStart : newTime,
          LastClientId = client != null
                         ? client.Id : lastFact != null
                         ? lastFact.LastClientId : (Guid?)null
        };

      var endStates = new[] { TaskState.Finished, TaskState.Failed, TaskState.Aborted };
      factTaskDao.Save(
        from x in newTasks.ToList()
        let stateLogsLinkedList = new LinkedList<StateLog>(x.StateLogs.OrderBy(y => y.DateTime))
        let lastStateLog = stateLogsLinkedList.OrderByDescending(y => y.DateTime).FirstOrDefault(sl => sl.Exception != null)
        select new FactTask {
          TaskId = x.TaskId,
          JobId = x.JobId,
          DimTimeStart = x.DimTimeStart,
          DimTimeEnd = endStates.Contains(x.State) ? newTime : null,
          LastClientId = x.LastClientId,
          Priority = x.Priority,
          CoresRequired = x.CoresRequired,
          MemoryRequired = x.MemoryRequired,
          NumCalculationRuns = stateLogsLinkedList.CountCalculationRuns(),
          NumRetries = stateLogsLinkedList.CountRetries(),
          TotalWaitingTime = stateLogsLinkedList.SumTotalTimeWhere(stateLog => stateLog.Value.State == TaskState.Waiting),
          TotalRuntime = stateLogsLinkedList.SumTotalTimeWhere(stateLog => stateLog.Value.State == TaskState.Calculating && stateLog.NextIs(y => y.State == TaskState.Transferring)),
          TotalTransferTime = stateLogsLinkedList.SumTotalTimeWhere(stateLog => stateLog.Value.State == TaskState.Transferring),
          TaskState = x.State,
          Exception = lastStateLog == null ? string.Empty : lastStateLog.Exception
        });
      factTaskDao.Delete(notFinishedFactTasks.Select(x => x.TaskId));
    }

    private string GetUserName(Guid userId) {
      try {
        var userManager = ServiceLocator.Instance.UserManager;
        var user = userManager.GetUserById(userId);
        return user != null ? user.UserName : UnknownUserName;
      }
      catch (Exception) {
        return UnknownUserName;
      }
    }
  }

  public static class StateLogLinkedListExtensions {
    public static int CountCalculationRuns(this LinkedList<StateLog> stateLogs) {
      return stateLogs.EnumerateNodes()
                      .Count(sl => sl.Value.State == TaskState.Calculating && sl.NextIs(nsl => nsl.State == TaskState.Transferring));
    }

    public static int CountRetries(this LinkedList<StateLog> stateLogs) {
      return stateLogs.EnumerateNodes()
                      .Count(sl => sl.Value.State == TaskState.Calculating && sl.Next != null && sl.NextIs(nsl => nsl.State != TaskState.Transferring));
    }

    public static double SumTotalTimeWhere(this LinkedList<StateLog> stateLogs, Predicate<LinkedListNode<StateLog>> predicate) {
      return stateLogs.EnumerateNodes()
                      .Where(stateLog => predicate(stateLog))
                      .Sum(stateLog => stateLog.Next != null ? (stateLog.Next.Value.DateTime - stateLog.Value.DateTime).TotalMinutes : 0.0);
    }
  }

  public static class LinkedListExtensions {
    public static IEnumerable<LinkedListNode<T>> EnumerateNodes<T>(this LinkedList<T> list) {
      var node = list.First;
      while (node != null) {
        yield return node;
        node = node.Next;
      }
    }

    public static bool NextIs<T>(this LinkedListNode<T> node, Predicate<T> predicate) {
      return node.Next != null && predicate(node.Next.Value);
    }
  }
}