Merge pull request baidu#1 from leoYY/galaxy2.0

Galaxy2.0

src/flags.cc

@@ -42,3 +42,4 @@ DEFINE_string(gce_initd_dump_file, "initd_checkpoint_file", "Initd Checkpoint Fi
 DEFINE_string(gce_initd_port, "8765", "gce initd listen port");
 DEFINE_string(gce_gced_port, "8766", "gce initd listen port");
 DEFINE_string(gce_initd_bin, "./initd", "initd bin path");
+DEFINE_string(gce_work_dir, "./workdir", "gce workdir path");

src/gce/cgroups.cc

@@ -0,0 +1,105 @@
+// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "gce/cgroups.h"
+
+#include <unistd.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <boost/lexical_cast.hpp>
+#include <boost/algorithm/string.hpp>
+#include <boost/algorithm/string/split.hpp>
+
+namespace baidu {
+namespace galaxy {
+namespace cgroups {
+
+bool GetPidsFromCgroup(const std::string& hierarchy,
+                       const std::string& cgroup,
+                       std::vector<std::string>* pids) {
+    if (pids == NULL) {
+        return false; 
+    }
+    std::string value;
+    if (0 != Read(hierarchy, cgroup, "cgroup.proc", &value)) {
+        return false; 
+    }
+
+    LOG(DEBUG, "get from cgroup.proc %s", value.c_str());
+    boost::split(*pids, value, boost::is_any_of("\n"));    
+    LOG(DEBUG, "get from cgroup.proc pids %u", pids->size());
+    return true;
+}
+
+bool AttachPid(const std::string& hierarchy,
+               const std::string& cgroup,
+               int pid) {
+    if (0 == Write(
+                hierarchy, 
+                cgroup, 
+                "tasks", 
+                boost::lexical_cast<std::string>(pid))) {
+        return true; 
+    }
+    return false;
+}
+
+int Write(const std::string& hierarchy,
+          const std::string& cgroup,
+          const std::string& control_file,
+          const std::string& value) {                        
+    std::string file_path = hierarchy + "/" 
+        + cgroup + "/" + control_file;
+    FILE* fd = ::fopen(file_path.c_str(), "we");
+    if (fd == NULL) {
+        return -1;
+    }
+
+    int ret = fprintf(fd, "%s", value.c_str()); 
+    fclose();
+    if (ret <= 0) {
+        return -1; 
+    }
+    return 0;
+}
+
+int Read(const std::string& hierarchy,
+         const std::string& cgroup,
+         const std::string& control_file,
+         std::string* value) {
+    if (value == NULL) {
+        return -1; 
+    }    
+    std::string file_path = hierarchy + "/" 
+        + cgroup + "/" + control_file;
+    FILE* fin = ::fopen(file_path.c_str(), "re"); 
+    if (NULL == fin) {
+        return -1; 
+    }
+
+    const int TEMP_BUF_SIZE = 30;
+    char temp_read_buffer[TEMP_BUF_SIZE];
+    while (::feof(fin) == 0) {
+        size_t readlen = ::fread((void*)temp_read_buffer, 
+                                 sizeof(char), 
+                                 TEMP_BUF_SIZE, fin); 
+        if (readlen == 0) {
+            break; 
+        }
+        value->append(temp_read_buffer, readlen);
+    }
+
+    if (::ferror(fin) != 0) {
+        ::fclose(fin);
+        return -1;
+    }
+    ::fclose(fin);
+    return 0;
+}
+
+
+}   // ending namespace cgroups
+}   // ending namespace galaxy
+}   // ending namespace baidu

src/gce/cgroups.h

@@ -0,0 +1,42 @@
+// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef _CGROUPS_H_
+#define _CGROUPS_H_
+
+#include <string>
+#include <vector>
+
+namespace baidu {
+namespace galaxy {
+namespace cgroups {
+
+bool GetPidsFromCgroup(const std::string& hierarchy, 
+                       const std::string& cgroup, 
+                       std::vector<std::string>* pids);
+
+bool AttachPid(const std::string& hierarchy,
+               const std::string& cgroup,
+               int pid);
+
+// hierarchy path for cgroups root. example : /cgroups/subsystem/
+// cgroup relative path for hierarchy. example ./xxxxxx/
+// control file. example cpu.share
+// value    
+int Write(const std::string& hierarchy, 
+          const std::string& cgroup, 
+          const std::string& control_file, 
+          const std::string& value);
+
+int Read(const std::string& hierarchy,
+         const std::string& cgroup,
+         const std::string& control_file,
+         std::string* value);
+
+}   // ending namespace cgroups
+}   // ending namespace galaxy
+}   // ending namespace baidu
+
+#endif
+

src/gce/task_manager.cc

@@ -20,21 +20,96 @@
 DECLARE_string(gce_cgroup_root);
 DECLARE_string(gce_support_subsystems);
 DECLARE_int64(gce_initd_zombie_check_interval);
+DECLARE_string(gce_work_dir);
 
 namespace baidu {
 namespace galaxy {
 
-TaskManager::TaskManager() {
+TaskManager::TaskManager() : 
+    tasks_mutex_(),
+    tasks_(),
+    background_thread_(1), 
+    cgroup_root_(FLAGS_gce_cgroup_root),
+    support_subsystems_() {
+    // init resource collector engine
 }
 
 TaskManager::~TaskManager() {
 }
 
-int TaskManager::CreatePodTasks(const PodDescriptor& pod) {
+int TaskManager::Init() {
+    // TODO create susystem deal
     return 0;
 }
 
-int TaskManager::DeletePodTasks(const PodDescriptor& pod) {
+int TaskManager::CreatePodTasks(const std::string& podid, const PodDescriptor& pod) {
+    MutexLock scope_lock(&tasks_mutex_);
+    for (int i = 0; i < pod.tasks_size(); i++) {
+        const TaskDescriptor& task = pod.tasks(i);  
+        TaskInfo* task_info = new TaskInfo();
+        task_info->task_id = GenerateTaskId(podid);
+        task_info->pod_id = podid;
+        task_info->desc.CopyFrom(task);
+        task_info->status.set_state(kPodPending);
+        task_info->millicores = task.requirement().millicores();
+        tasks_[task_info->task_id] = task_info;
+        // 1. prepare workspace 
+        if (PrepareWorkspace(task_info) != 0) {
+            LOG(WARNING, "prepare task %s workspace failed", 
+                    task_info->task_id.c_str()); 
+            return -1;
+        }
+        // 2. prepare cgroup 
+        if (PrepareCgroupEnv(task_info) != 0) {
+            LOG(WARNING, "prepare task %s cgroup faield", 
+                    task_info->task_id.c_str()); 
+            return -1;
+        }
+        // 3. prepare mount path
+        if (PrepareVolumeEnv(task_info) != 0) {
+            LOG(WARNING, "prepare task %s volume env failed",
+                    task_info->task_id.c_str()); 
+            return -1;
+        }
+        LOG(INFO, "prepare task %s success", 
+                        task_info->task_id.c_str());
+    }
+    return 0;
+}
+
+int TaskManager::DeletePodTasks() {
+    MutexLock scope_lock(&tasks_mutex_);
+    std::map<std::string, TaskInfo*>::iterator it = 
+        tasks_.begin();
+    for (; it != tasks_.end(); ++it) {
+        // 4. Terminate Task 
+        if (TerminateTask(it->second) != 0) {
+            LOG(WARNING, "terminate task %s failed", 
+                    it->first.c_str()); 
+            return -1;
+        }
+        // 3. clean mount path
+        if (CleanVolumeEnv(it->second) != 0) {
+            LOG(WARNING, "clean task %s volume failed",
+                    it->first.c_str()); 
+            return -1;
+        }
+        // 2. clean cgroup
+        if (CleanCgroupEnv(it->second) != 0) {
+            LOG(WARNING, "clean task %s cgroup failed", 
+                    it->first.c_str()); 
+            return -1;
+        }
+        // 1. clean workspace
+        if (CleanWorkspace(it->second) != 0) {
+            LOG(WARNING, "clean task %s workspace failed",
+                    it->first.c_str()); 
+            return -1;
+        }
+        delete it->second;
+        it->second = NULL;
+    }
+    tasks_.clear();  
     return 0;
 }
 
@@ -46,9 +121,66 @@ void TaskManager::LoopCheckTaskStatus() {
 }
 
 
-int Execute(const std::string& command) {
+int TaskManager::Execute(const std::string& command) {
+    return 0;
+}
+
+int TaskManager::RunTask(const TaskInfo* task_info) {
+    return 0;
+}
+
+int TaskManager::TerminateTask(const TaskInfo* task_info) {
+    return 0;
+}
+
+int TaskManager::PrepareWorkspace(TaskInfo* task) {
+    std::string workspace_root = FLAGS_gce_work_dir;
+    workspace_root.append("/");
+    workspace_root.append(task->pod_id);
+    if (!file::Mkdir(workspace_root)) {
+        LOG(WARNING, "mkdir workspace root failed"); 
+        return -1;
+    }
+
+    std::string task_workspace = workspace_root;
+    task_workspace.append("/");
+    task_workspace.append(task->task_id);
+    if (!file::Mkdir(task_workspace)) {
+        LOG(WARNING, "mkdir task workspace failed");
+        return -1;
+    }
+    task->task_workspace = task_workspace;
+    return 0;
+}
+
+int TaskManager::PrepareCgroupEnv(const TaskInfo* task) {
+    return 0;
+}
+
+int TaskManager::PrepareVolumeEnv(const TaskInfo* task) {
+    return 0;
+}
+
+int TaskManager::CleanWorkspace(const TaskInfo* task) {
+    if (file::IsExists(task->task_workspace)
+            && !file::Remove(task->task_workspace)) {
+        LOG(WARNING, "Remove task %s workspace failed",
+                task->task_id.c_str());
+        return -1;
+    }
     return 0;
-
+}
+
+int TaskManager::CleanCgroupEnv(const TaskInfo* task) {
+    return 0;
+}
+
+int TaskManager::CleanVolumeEnv(const TaskInfo* task) {
+    return 0;
+}
+
+std::string TaskManager::GenerateTaskId(const std::string& podid) {
+    return "";
 }
 
 } // ending namespace galaxy