Line data Source code
1 : /*
2 : * Copyright (c) 2014 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "oper/instance_manager.h"
7 :
8 : #include <boost/bind/bind.hpp>
9 : #include <boost/functional/hash.hpp>
10 : #include <boost/filesystem.hpp>
11 : #include <boost/tokenizer.hpp>
12 : #include <sys/wait.h>
13 : #include "cmn/agent.h"
14 : #include "db/db.h"
15 : #include "io/event_manager.h"
16 : #include "oper/instance_task.h"
17 : #include "oper/operdb_init.h"
18 : #include "oper/service_instance.h"
19 : #include "oper/vm.h"
20 : #include "oper/docker_instance_adapter.h"
21 : #include "oper/netns_instance_adapter.h"
22 : #ifdef WITH_LIBVIRT
23 : #include "oper/libvirt_instance_adapter.h"
24 : #endif
25 : #include "base/util.h"
26 :
27 : using namespace boost::placeholders;
28 :
29 : using boost::uuids::uuid;
30 : SandeshTraceBufferPtr InstanceManagerTraceBuf(
31 : SandeshTraceBufferCreate("InstanceManager", 1000));
32 :
33 : static const char loadbalancer_config_path_default[] =
34 : "/var/lib/contrail/loadbalancer/";
35 : static const char namespace_store_path_default[] =
36 : "/var/run/netns";
37 : static const char namespace_prefix[] = "vrouter-";
38 :
39 : class InstanceManager::NamespaceStaleCleaner {
40 : public:
41 19 : NamespaceStaleCleaner(Agent *agent, InstanceManager *manager)
42 19 : : agent_(agent), manager_(manager) {
43 19 : }
44 :
45 19 : void CleanStaleEntries() {
46 : namespace fs = boost::filesystem;
47 :
48 : //Read all the Namepaces in the system
49 19 : fs::path ns(manager_->namespace_store_path_);
50 19 : if ( !fs::exists(ns) || !fs::is_directory(ns)) {
51 19 : return;
52 : }
53 :
54 : typedef boost::tokenizer<boost::char_separator<char> > tokenizer;
55 0 : boost::char_separator<char> slash_sep("/");
56 0 : boost::char_separator<char> colon_sep(":");
57 0 : fs::directory_iterator end_iter;
58 0 : for(fs::directory_iterator iter(ns); iter != end_iter; iter++) {
59 :
60 : // Get to the name of namespace by removing complete path
61 0 : tokenizer tokens(iter->path().string(), slash_sep);
62 0 : std::string ns_name;
63 0 : for(tokenizer::iterator it=tokens.begin(); it!=tokens.end(); it++){
64 0 : ns_name = *it;
65 0 : }
66 :
67 : //We are interested only in namespaces starting with a given
68 : //prefix
69 : std::size_t vrouter_found;
70 0 : vrouter_found = ns_name.find(namespace_prefix);
71 0 : if (vrouter_found == std::string::npos) {
72 0 : continue;
73 : }
74 :
75 : //Remove the standard prefix
76 0 : ns_name.replace(vrouter_found, strlen(namespace_prefix), "");
77 :
78 : //Namespace might have a ":". Extract both left and right of
79 : //":" Left of ":" is the VM uuid. If not found in Agent's VM
80 : //DB, it can be deleted
81 0 : tokenizer tok(ns_name, colon_sep);
82 0 : boost::uuids::uuid vm_uuid = StringToUuid(*tok.begin());
83 0 : VmKey key(vm_uuid);
84 0 : if (agent_->vm_table()->Find(&key, true)) {
85 0 : continue;
86 : }
87 :
88 0 : ServiceInstance::Properties prop;
89 0 : prop.instance_id = vm_uuid;
90 0 : prop.service_type = ServiceInstance::SourceNAT;
91 0 : tokenizer::iterator next_tok = ++(tok.begin());
92 : //Loadbalancer namespace
93 0 : if (next_tok != tok.end()) {
94 0 : prop.loadbalancer_id = *next_tok;
95 0 : prop.service_type = ServiceInstance::LoadBalancer;
96 : }
97 :
98 : //Delete Namespace
99 0 : manager_->StopStaleNetNS(prop);
100 :
101 : //If Loadbalncer, delete the config files as well
102 0 : if (prop.service_type == ServiceInstance::LoadBalancer) {
103 : //Delete the complete directory
104 0 : std::stringstream pathgen;
105 0 : pathgen << manager_->loadbalancer_config_path_
106 0 : << prop.loadbalancer_id << ".conf";
107 :
108 0 : boost::system::error_code error;
109 0 : if (fs::exists(pathgen.str())) {
110 0 : fs::remove(pathgen.str(), error);
111 0 : if (error) {
112 0 : std::stringstream ss;
113 0 : ss << "Stale loadbalancer cfg fle delete error ";
114 0 : ss << error.message();
115 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
116 0 : }
117 : }
118 0 : }
119 0 : }
120 19 : }
121 :
122 : private:
123 : Agent *agent_;
124 : InstanceManager *manager_;
125 : };
126 :
127 2 : InstanceManager::~InstanceManager() {
128 2 : TimerManager::DeleteTimer(stale_timer_);
129 2 : STLDeleteValues(&adapters_);
130 2 : }
131 :
132 2 : InstanceManager::InstanceManager(Agent *agent)
133 2 : : si_listener_(DBTableBase::kInvalidId),
134 2 : netns_timeout_(-1),
135 2 : work_queue_(TaskScheduler::GetInstance()->GetTaskId(INSTANCE_MANAGER_TASK_NAME), 0,
136 : boost::bind(&InstanceManager::DequeueEvent, this, _1)),
137 2 : loadbalancer_config_path_(loadbalancer_config_path_default),
138 2 : namespace_store_path_(namespace_store_path_default),
139 2 : stale_timer_interval_(5 * 60 * 1000),
140 2 : stale_timer_(TimerManager::CreateTimer(*(agent->event_manager()->io_service()),
141 : "NameSpaceStaleTimer", TaskScheduler::GetInstance()->
142 6 : GetTaskId(INSTANCE_MANAGER_TASK_NAME), 0)), agent_(agent) {
143 2 : if (agent->isMockMode()) {
144 0 : loadbalancer_config_path_ = "/tmp/" + agent->AgentGUID()
145 0 : + loadbalancer_config_path_default;
146 0 : namespace_store_path_ = "/tmp/" + agent->AgentGUID()
147 0 : + namespace_store_path_default;
148 : }
149 2 : work_queue_.set_name("Instance Manager");
150 :
151 2 : }
152 :
153 2 : void InstanceManager::Initialize(DB *database, const std::string &netns_cmd,
154 : const std::string &docker_cmd,
155 : const int netns_workers,
156 : const int netns_timeout) {
157 2 : DBTableBase *si_table = agent_->service_instance_table();
158 2 : assert(si_table);
159 2 : si_listener_ = si_table->Register(
160 : boost::bind(&InstanceManager::EventObserver, this, _1, _2));
161 :
162 2 : netns_cmd_ = netns_cmd;
163 2 : if (netns_cmd_.length() == 0) {
164 2 : LOG(ERROR, "NetNS path for network namespace command not specified "
165 : "in the config file, the namespaces won't be started");
166 : }
167 2 : if (docker_cmd.length() == 0) {
168 2 : LOG(ERROR, "Path for Docker starter command not specified "
169 : "in the config file, the Docker instances won't be started");
170 : }
171 :
172 2 : std::stringstream pathgen;
173 2 : pathgen << loadbalancer_config_path_;
174 2 : boost::filesystem::path dir(pathgen.str());
175 2 : boost::system::error_code error;
176 2 : boost::filesystem::create_directories(dir, error);
177 2 : if (error) {
178 0 : LOG(ERROR, "Falied to create Loadbalancer Directory " << pathgen.str());
179 : }
180 :
181 2 : adapters_.push_back(new DockerInstanceAdapter(docker_cmd, agent_));
182 2 : adapters_.push_back(new NetNSInstanceAdapter(netns_cmd,
183 2 : loadbalancer_config_path_, agent_));
184 : #ifdef WITH_LIBVIRT
185 : adapters_.push_back(new LibvirtInstanceAdapter(agent_,
186 : "qemu:///system"));
187 : #endif
188 :
189 2 : netns_timeout_ = kTimeoutDefault;
190 2 : if (netns_timeout >= 1) {
191 0 : netns_timeout_ = netns_timeout;
192 : }
193 :
194 2 : netns_reattempts_ = kReattemptsDefault;
195 :
196 2 : int workers = kWorkersDefault;
197 2 : if (netns_workers > 0) {
198 0 : workers = netns_workers;
199 : }
200 :
201 2 : task_queues_.resize(workers);
202 2 : for (std::vector<InstanceTaskQueue *>::iterator iter = task_queues_.begin();
203 4 : iter != task_queues_.end(); ++iter) {
204 : InstanceTaskQueue *task_queue =
205 2 : new InstanceTaskQueue(agent_->event_manager());
206 2 : assert(task_queue);
207 2 : task_queue->set_on_timeout_cb(
208 : boost::bind(&InstanceManager::OnTaskTimeout,
209 : this, _1));
210 2 : *iter = task_queue;
211 : }
212 2 : stale_timer_->Start(StaleTimerInterval(),
213 : boost::bind(&InstanceManager::StaleTimeout, this));
214 :
215 2 : }
216 :
217 0 : void InstanceManager::SetNetNSCmd(const std::string &netns_cmd) {
218 0 : ServiceInstance::Properties prop;
219 0 : prop.virtualization_type =
220 : ServiceInstance::ServiceInstance::NetworkNamespace;
221 : NetNSInstanceAdapter *adapter = static_cast<NetNSInstanceAdapter
222 0 : *>(FindApplicableAdapter(prop));
223 0 : if (adapter)
224 0 : adapter->set_cmd(netns_cmd);
225 0 : }
226 :
227 0 : void InstanceManager::SetStaleTimerInterval(int minutes) {
228 0 : stale_timer_interval_ = minutes * 60 * 1000;
229 0 : }
230 :
231 0 : void InstanceManager::OnTaskTimeout(InstanceTaskQueue *task_queue) {
232 0 : InstanceManagerChildEvent event;
233 0 : event.type = OnTaskTimeoutEvent;
234 0 : event.task_queue = task_queue;
235 :
236 0 : work_queue_.Enqueue(event);
237 0 : }
238 :
239 0 : void InstanceManager::OnTaskTimeoutEventHandler(InstanceManagerChildEvent event) {
240 0 : std::stringstream ss;
241 0 : ss << "TaskTimeOut for the TaskQ " << event.task_queue;
242 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
243 :
244 0 : ScheduleNextTask(event.task_queue);
245 0 : }
246 :
247 0 : void InstanceManager::OnErrorEventHandler(InstanceManagerChildEvent event) {
248 0 : ServiceInstance *svc_instance = GetSvcInstance(event.task);
249 0 : if (!svc_instance) {
250 0 : return;
251 : }
252 :
253 0 : std::stringstream ss;
254 0 : ss << "Error for the Task " << event.task << " " << event.errors;
255 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
256 :
257 0 : InstanceState *state = GetState(svc_instance);
258 0 : if (state != NULL) {
259 0 : state->set_errors(event.errors);
260 : }
261 0 : }
262 :
263 0 : void InstanceManager::OnExitEventHandler(InstanceManagerChildEvent event) {
264 0 : ServiceInstance *svc_instance = GetSvcInstance(event.task);
265 0 : if (!svc_instance) {
266 0 : return;
267 : }
268 :
269 0 : std::stringstream ss;
270 0 : ss << "Exit event for the Task " << event.task;
271 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
272 :
273 0 : UpdateStateStatusType(event);
274 0 : for (std::vector<InstanceTaskQueue *>::iterator iter =
275 0 : task_queues_.begin();
276 0 : iter != task_queues_.end(); ++iter) {
277 0 : InstanceTaskQueue *task_queue = *iter;
278 0 : if (!task_queue->Empty()) {
279 0 : if (task_queue->Front() == event.task) {
280 0 : task_queue->Pop();
281 0 : delete event.task;
282 0 : task_queue->StopTimer();
283 0 : DeleteState(svc_instance);
284 0 : ScheduleNextTask(task_queue);
285 0 : return;
286 : }
287 : }
288 : }
289 :
290 0 : }
291 :
292 0 : bool InstanceManager::DequeueEvent(InstanceManagerChildEvent event) {
293 0 : if (event.type == OnErrorEvent) {
294 0 : OnErrorEventHandler(event);
295 0 : } else if (event.type == OnTaskTimeoutEvent) {
296 0 : OnTaskTimeoutEventHandler(event);
297 0 : } else if (event.type == OnExitEvent) {
298 0 : OnExitEventHandler(event);
299 : }
300 :
301 0 : return true;
302 : }
303 :
304 0 : void InstanceManager::UpdateStateStatusType(InstanceManagerChildEvent event) {
305 0 : ServiceInstance* svc_instance = UnregisterSvcInstance(event.task);
306 0 : if (svc_instance) {
307 0 : InstanceState *state = GetState(svc_instance);
308 :
309 : // The below code might not really capture the errors that
310 : // occured in the child process. As we are relying on the pipe
311 : // status to identify child exit, even if child process ends up
312 : // in error state, pipe might return "success" because pipe is
313 : // closed without erros. But what we want to reflect is the
314 : // error of child process. If there is any error in the pipe
315 : // status, we show that as is. If not, if there is any error string
316 : // in error_, we mark the error status as -1.
317 : // pipe returning boost::system::errc::no_such_file_or_directory
318 : // error is considered as no pipe errors
319 :
320 0 : if (state != NULL) {
321 0 : int error_status = event.error_val;
322 0 : if (error_status ==
323 : boost::system::errc::no_such_file_or_directory) {
324 0 : error_status = 0;
325 : }
326 0 : if (!state->errors().empty()) {
327 0 : if (error_status == 0) {
328 0 : error_status = -1;
329 : }
330 : }
331 :
332 0 : state->set_status(error_status);
333 :
334 0 : if (error_status != 0) {
335 0 : if (state->status_type() != InstanceState::Timeout) {
336 0 : state->set_status_type(InstanceState::Error);
337 : }
338 0 : } else if (state->status_type() == InstanceState::Starting) {
339 0 : state->set_status_type(InstanceState::Started);
340 0 : } else if (state->status_type() == InstanceState::Stopping) {
341 0 : state->set_status_type(InstanceState::Stopped);
342 : }
343 :
344 0 : std::stringstream ss;
345 0 : ss << "For the task " << event.task << " error status " <<
346 0 : error_status << " status type " << state->status_type();
347 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
348 0 : }
349 : }
350 0 : }
351 :
352 0 : InstanceState *InstanceManager::GetState(ServiceInstance *svc_instance) const {
353 : return static_cast<InstanceState *>(
354 0 : svc_instance->GetState(agent_->service_instance_table(),
355 0 : si_listener_));
356 : }
357 :
358 0 : InstanceState *InstanceManager::GetState(InstanceTask* task) const {
359 0 : ServiceInstance* svc_instance = GetSvcInstance(task);
360 0 : if (svc_instance) {
361 0 : InstanceState *state = GetState(svc_instance);
362 0 : return state;
363 : }
364 0 : return NULL;
365 : }
366 :
367 0 : void InstanceManager::SetState(ServiceInstance *svc_instance,
368 : InstanceState *state) {
369 0 : svc_instance->SetState(agent_->service_instance_table(),
370 : si_listener_,state);
371 0 : }
372 :
373 0 : void InstanceManager::ClearState(ServiceInstance *svc_instance) {
374 0 : svc_instance->ClearState(agent_->service_instance_table(), si_listener_);
375 0 : }
376 :
377 0 : bool InstanceManager::DeleteState(ServiceInstance *svc_instance) {
378 :
379 0 : if (!svc_instance || !svc_instance->IsDeleted()) {
380 0 : return false;
381 : }
382 :
383 0 : InstanceState *state = GetState(svc_instance);
384 0 : if (state && !state->tasks_running()) {
385 0 : ClearState(svc_instance);
386 0 : delete state;
387 0 : ClearLastCmdType(svc_instance);
388 0 : return true;
389 : }
390 :
391 0 : return false;
392 : }
393 :
394 :
395 2 : void InstanceManager::StateClear() {
396 : DBTablePartition *partition = static_cast<DBTablePartition *>(
397 2 : agent_->service_instance_table()->GetTablePartition(0));
398 :
399 2 : if (!partition)
400 0 : return;
401 :
402 2 : DBEntryBase *next = NULL;
403 2 : for (DBEntryBase *entry = partition->GetFirst(); entry; entry = next) {
404 0 : next = partition->GetNext(entry);
405 : DBState *state =
406 0 : entry->GetState(agent_->service_instance_table(), si_listener_);
407 0 : if (state != NULL) {
408 0 : entry->ClearState(agent_->service_instance_table(), si_listener_);
409 0 : delete state;
410 0 : ClearLastCmdType(static_cast<ServiceInstance *>(entry));
411 : }
412 : }
413 : }
414 :
415 2 : void InstanceManager::Terminate() {
416 2 : StateClear();
417 2 : agent_->service_instance_table()->Unregister(si_listener_);
418 2 : agent_->service_instance_table()->Clear();
419 :
420 : InstanceTaskQueue *task_queue;
421 2 : for (std::vector<InstanceTaskQueue *>::iterator iter = task_queues_.begin();
422 4 : iter != task_queues_.end(); ++iter) {
423 2 : if ((task_queue = *iter) == NULL) {
424 0 : continue;
425 : }
426 2 : task_queue->Clear();
427 :
428 2 : delete task_queue;
429 : }
430 2 : work_queue_.Shutdown();
431 2 : }
432 :
433 0 : void InstanceManager::Enqueue(InstanceTask *task,
434 : const boost::uuids::uuid &uuid) {
435 0 : std::stringstream ss;
436 0 : ss << uuid;
437 0 : InstanceTaskQueue *task_queue = GetTaskQueue(ss.str());
438 0 : task_queue->Push(task);
439 0 : ScheduleNextTask(task_queue);
440 0 : }
441 :
442 0 : InstanceTaskQueue *InstanceManager::GetTaskQueue(const std::string &str) {
443 : boost::hash<std::string> hash;
444 0 : int index = hash(str) % task_queues_.size();
445 0 : return task_queues_[index];
446 : }
447 :
448 : //After Run(), if child process is running, we keep the task status as
449 : //"Starting" or "Stopping". We start a timer to track TaskTimeout time.
450 : //if process is not runnning, we verify how many times we already attempted
451 : //to run. If netns_reattempts_ are already crossed, we return false,
452 : // so that caller deletes the task without running any further.
453 : //If required reattempts are not done, we start a timer and return true
454 : // so that same task is run again after the timeout. The task status is set to
455 : //"reattempt" to track reattempt case
456 0 : bool InstanceManager::StartTask(InstanceTaskQueue *task_queue,
457 : InstanceTask *task) {
458 :
459 :
460 0 : InstanceState *state = GetState(task);
461 0 : if (state) {
462 0 : state->reset_errors();
463 : }
464 :
465 : pid_t pid;
466 0 : bool status = task->Run();
467 :
468 0 : std::stringstream ss;
469 0 : ss << "Run status for the task " << task << " " << status;
470 0 : ss << " With running status " << task->is_running();
471 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
472 :
473 0 : if (status || task->is_running()) {
474 0 : pid = task->pid();
475 0 : if (state != NULL) {
476 0 : state->set_pid(pid);
477 0 : state->set_cmd(task->cmd());
478 0 : if (task->cmd_type() == Start) {
479 0 : state->set_status_type(InstanceState::Starting);
480 : } else {
481 0 : state->set_status_type(InstanceState::Stopping);
482 : }
483 : }
484 : } else {
485 :
486 0 : ss.str(std::string());
487 0 : ss << "Run failure for the task " << task << " attempt " << task->reattempts();
488 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
489 :
490 0 : if (state) {
491 0 : state->set_status_type(InstanceState::Reattempt);
492 0 : state->set_cmd(task->cmd());
493 : }
494 0 : if (task->incr_reattempts() > netns_reattempts_) {
495 0 : ss.str(std::string());
496 0 : ss << "Run failure for the task " << task << " attempts exceeded";
497 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
498 0 : return false;
499 : }
500 : }
501 :
502 0 : task_queue->StartTimer(netns_timeout_ * 1000);
503 :
504 0 : return true;
505 0 : }
506 :
507 : //If Starting the task succeds we wait for another event on that task.
508 : //If not the task is removed from the front of the queue and is delted.
509 0 : void InstanceManager::ScheduleNextTask(InstanceTaskQueue *task_queue) {
510 0 : while (!task_queue->Empty()) {
511 :
512 0 : InstanceTask *task = task_queue->Front();
513 0 : InstanceState *state = GetState(task);
514 :
515 0 : if (!task->is_running()) {
516 0 : bool status = StartTask(task_queue, task);
517 0 : if (status) {
518 0 : return;
519 : }
520 : } else {
521 0 : int delay = time(NULL) - task->start_time();
522 0 : if (delay < netns_timeout_) {
523 0 : return;
524 : }
525 0 : if (state) {
526 0 : state->set_status_type(InstanceState::Timeout);
527 : }
528 :
529 0 : std::stringstream ss;
530 0 : ss << "Timeout for the Task " << task << " delay " << delay;
531 0 : ss << " netns timeout " << netns_timeout_ << " ";
532 0 : ss << task->cmd();
533 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
534 :
535 0 : if (delay >= (netns_timeout_ * 2)) {
536 0 : task->Terminate();
537 0 : if (task->IsSetup())
538 0 : return;
539 : } else {
540 0 : task->Stop();
541 0 : return;
542 : }
543 0 : }
544 :
545 0 : task_queue->StopTimer();
546 0 : task_queue->Pop();
547 :
548 0 : ServiceInstance* svc_instance = GetSvcInstance(task);
549 0 : if (state && svc_instance)
550 0 : state->decr_tasks_running();
551 :
552 0 : task_svc_instances_.erase(task);
553 :
554 0 : std::stringstream ss;
555 0 : ss << "Delete of the Task " << task;
556 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
557 :
558 0 : DeleteState(svc_instance);
559 :
560 0 : delete task;
561 0 : }
562 : }
563 :
564 0 : ServiceInstance *InstanceManager::GetSvcInstance(InstanceTask *task) const {
565 : TaskSvcMap::const_iterator iter =
566 0 : task_svc_instances_.find(task);
567 0 : if (iter != task_svc_instances_.end()) {
568 0 : return iter->second;
569 : }
570 0 : return NULL;
571 : }
572 :
573 0 : void InstanceManager::RegisterSvcInstance(InstanceTask *task,
574 : ServiceInstance *svc_instance) {
575 : pair<TaskSvcMap::iterator, bool> result =
576 0 : task_svc_instances_.insert(std::make_pair(task, svc_instance));
577 0 : assert(result.second);
578 :
579 0 : InstanceState *state = GetState(svc_instance);
580 0 : assert(state);
581 0 : state->incr_tasks_running();
582 0 : }
583 :
584 0 : ServiceInstance *InstanceManager::UnregisterSvcInstance(InstanceTask *task) {
585 0 : for (TaskSvcMap::iterator iter =
586 0 : task_svc_instances_.begin();
587 0 : iter != task_svc_instances_.end(); ++iter) {
588 0 : if (task == iter->first) {
589 0 : ServiceInstance *svc_instance = iter->second;
590 0 : InstanceState *state = GetState(svc_instance);
591 0 : assert(state);
592 0 : state->decr_tasks_running();
593 0 : task_svc_instances_.erase(iter);
594 0 : return svc_instance;
595 : }
596 : }
597 :
598 0 : return NULL;
599 : }
600 :
601 0 : void InstanceManager::UnregisterSvcInstance(ServiceInstance *svc_instance) {
602 :
603 0 : InstanceState *state = GetState(svc_instance);
604 0 : assert(state);
605 :
606 : TaskSvcMap::iterator iter =
607 0 : task_svc_instances_.begin();
608 0 : while(iter != task_svc_instances_.end()) {
609 0 : if (svc_instance == iter->second) {
610 0 : task_svc_instances_.erase(iter++);
611 0 : state->decr_tasks_running();
612 : } else {
613 0 : ++iter;
614 : }
615 : }
616 0 : }
617 :
618 0 : InstanceManagerAdapter* InstanceManager::FindApplicableAdapter(const ServiceInstance::Properties &props) {
619 0 : for (std::vector<InstanceManagerAdapter *>::iterator iter = adapters_.begin();
620 0 : iter != adapters_.end(); ++iter) {
621 0 : InstanceManagerAdapter *adapter = *iter;
622 0 : if (adapter != NULL && adapter->isApplicable(props)) {
623 0 : return adapter;
624 : }
625 : }
626 0 : return NULL;
627 : }
628 :
629 0 : void InstanceManager::StartServiceInstance(ServiceInstance *svc_instance,
630 : InstanceState *state, bool update) {
631 0 : const ServiceInstance::Properties &props = svc_instance->properties();
632 0 : InstanceManagerAdapter *adapter = this->FindApplicableAdapter(props);
633 0 : std::stringstream ss;
634 0 : if (adapter != NULL) {
635 0 : InstanceTask *task = adapter->CreateStartTask(props, update);
636 0 : if (task != NULL) {
637 0 : ss << "Starting the Task " << task << " " << task->cmd();
638 0 : ss << " for " << props.instance_id;
639 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
640 0 : task->set_on_data_cb(boost::bind(&InstanceManager::OnError,
641 : this, _1, _2));
642 0 : task->set_on_exit_cb(boost::bind(&InstanceManager::OnExit,
643 : this, _1, _2));
644 0 : state->set_properties(props);
645 0 : RegisterSvcInstance(task, svc_instance);
646 0 : std::stringstream info;
647 0 : Enqueue(task, props.instance_id);
648 0 : } else {
649 0 : ss << "Error Starting the Task for " << props.instance_id;
650 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
651 : }
652 : } else {
653 0 : ss << "Unknown virtualization type " << props.virtualization_type;
654 0 : ss << " for " << svc_instance->ToString();
655 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
656 : }
657 0 : }
658 :
659 :
660 0 : void InstanceManager::StopServiceInstance(ServiceInstance *svc_instance,
661 : InstanceState *state) {
662 0 : const ServiceInstance::Properties &props = state->properties();
663 0 : InstanceManagerAdapter *adapter = this->FindApplicableAdapter(props);
664 0 : std::stringstream ss;
665 0 : if (adapter != NULL) {
666 0 : InstanceTask *task = adapter->CreateStopTask(props);
667 0 : if (task != NULL) {
668 0 : ss << "Stopping the Task " << task << " " << task->cmd();
669 0 : ss << " for " << props.instance_id;
670 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
671 0 : task->set_on_data_cb(boost::bind(&InstanceManager::OnError,
672 : this, _1, _2));
673 0 : task->set_on_exit_cb(boost::bind(&InstanceManager::OnExit,
674 : this, _1, _2));
675 0 : RegisterSvcInstance(task, svc_instance);
676 0 : std::stringstream info;
677 0 : Enqueue(task, props.instance_id);
678 0 : } else {
679 0 : std::stringstream ss;
680 0 : ss << "Error Stopping the Task for " << props.instance_id;
681 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
682 0 : }
683 : } else {
684 0 : ss << "Unknown virtualization type " << props.virtualization_type;
685 0 : ss << " for " << svc_instance->ToString();
686 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
687 : }
688 0 : }
689 :
690 0 : void InstanceManager::OnError(InstanceTask *task,
691 : const std::string errors) {
692 :
693 0 : InstanceManagerChildEvent event;
694 0 : event.type = OnErrorEvent;
695 0 : event.task = task;
696 0 : event.errors = errors;
697 :
698 0 : work_queue_.Enqueue(event);
699 0 : }
700 :
701 0 : void InstanceManager::OnExit(InstanceTask *task,
702 : const boost::system::error_code &ec) {
703 :
704 0 : InstanceManagerChildEvent event;
705 0 : event.type = OnExitEvent;
706 0 : event.task = task;
707 0 : event.error_val = ec.value();
708 :
709 0 : work_queue_.Enqueue(event);
710 0 : }
711 :
712 0 : void InstanceManager::StopStaleNetNS(ServiceInstance::Properties &props) {
713 0 : std::stringstream cmd_str;
714 :
715 0 : if (netns_cmd_.length() == 0) {
716 0 : return;
717 : }
718 0 : cmd_str << netns_cmd_ << " destroy";
719 :
720 0 : cmd_str << " " << props.ServiceTypeString();
721 0 : cmd_str << " " << UuidToString(props.instance_id);
722 0 : cmd_str << " " << UuidToString(boost::uuids::nil_uuid());
723 0 : cmd_str << " " << UuidToString(boost::uuids::nil_uuid());
724 0 : if (props.service_type == ServiceInstance::LoadBalancer) {
725 0 : if (props.loadbalancer_id.empty()) {
726 0 : LOG(ERROR, "loadbalancer id is missing for service instance: "
727 : << UuidToString(props.instance_id));
728 0 : return;
729 : }
730 0 : cmd_str << " --loadbalancer-id " << props.loadbalancer_id;
731 : }
732 :
733 0 : std::string cmd = cmd_str.str();
734 0 : std::vector<std::string> argv;
735 0 : boost::split(argv, cmd, boost::is_any_of(" "), boost::token_compress_on);
736 0 : std::vector<const char *> c_argv(argv.size() + 1);
737 0 : for (std::size_t i = 0; i != argv.size(); ++i) {
738 0 : c_argv[i] = argv[i].c_str();
739 : }
740 :
741 0 : std::stringstream ss;
742 0 : ss << "StaleNetNS " << cmd;
743 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
744 :
745 0 : pid_t pid = vfork();
746 0 : if (pid == 0) {
747 0 : CloseTaskFds();
748 0 : execvp(c_argv[0], (char **) c_argv.data());
749 0 : perror("execvp");
750 :
751 0 : _exit(127);
752 : }
753 0 : }
754 :
755 0 : void InstanceManager::SetLastCmdType(ServiceInstance *svc_instance,
756 : int last_cmd_type) {
757 0 : std::string uuid = UuidToString(svc_instance->uuid());
758 : std::map<std::string, int>::iterator iter =
759 0 : last_cmd_types_.find(uuid);
760 0 : if (iter != last_cmd_types_.end()) {
761 0 : iter->second = last_cmd_type;
762 : } else {
763 0 : last_cmd_types_.insert(std::make_pair(uuid, last_cmd_type));
764 : }
765 0 : }
766 :
767 0 : int InstanceManager::GetLastCmdType(ServiceInstance *svc_instance) const {
768 0 : std::string uuid = UuidToString(svc_instance->uuid());
769 : std::map<std::string, int>::const_iterator iter =
770 0 : last_cmd_types_.find(uuid);
771 0 : if (iter != last_cmd_types_.end()) {
772 0 : return iter->second;
773 : }
774 :
775 0 : return 0;
776 0 : }
777 :
778 0 : void InstanceManager::ClearLastCmdType(ServiceInstance *svc_instance) {
779 0 : std::string uuid = UuidToString(svc_instance->uuid());
780 : std::map<std::string, int>::iterator iter =
781 0 : last_cmd_types_.find(uuid);
782 0 : if (iter != last_cmd_types_.end()) {
783 0 : last_cmd_types_.erase(iter);
784 : }
785 0 : }
786 :
787 0 : void InstanceManager::EventObserver(
788 : DBTablePartBase *db_part, DBEntryBase *entry) {
789 0 : ServiceInstance *svc_instance = static_cast<ServiceInstance *>(entry);
790 :
791 0 : InstanceState *state = GetState(svc_instance);
792 0 : if (svc_instance->IsDeleted()) {
793 0 : if (state) {
794 0 : if (GetLastCmdType(svc_instance) == Start) {
795 0 : StopServiceInstance(svc_instance, state);
796 0 : SetLastCmdType(svc_instance, Stop);
797 : }
798 0 : if (DeleteState(svc_instance)) {
799 0 : return;
800 : }
801 : }
802 0 : ClearLastCmdType(svc_instance);
803 : } else {
804 0 : if (state == NULL) {
805 0 : state = new InstanceState();
806 0 : SetState(svc_instance, state);
807 : }
808 :
809 0 : bool usable = svc_instance->IsUsable();
810 :
811 0 : std::stringstream ss;
812 0 : ss << "NetNS event notification for uuid: " << svc_instance->ToString();
813 0 : ss << (usable ? " usable" : " not usable");
814 0 : INSTANCE_MANAGER_TRACE(Trace, ss.str().c_str());
815 :
816 0 : if (!usable && GetLastCmdType(svc_instance) == Start) {
817 0 : StopServiceInstance(svc_instance, state);
818 0 : SetLastCmdType(svc_instance, Stop);
819 0 : } else if (usable) {
820 0 : if (GetLastCmdType(svc_instance) == Start && state->properties().CompareTo(
821 : svc_instance->properties()) != 0) {
822 0 : StartServiceInstance(svc_instance, state, true);
823 0 : } else if (GetLastCmdType(svc_instance) != Start) {
824 0 : StartServiceInstance(svc_instance, state, false);
825 0 : SetLastCmdType(svc_instance, Start);
826 : }
827 : }
828 0 : }
829 : }
830 :
831 19 : bool InstanceManager::StaleTimeout() {
832 :
833 19 : if (stale_cleaner_.get())
834 0 : return false;
835 19 : stale_cleaner_.reset(new NamespaceStaleCleaner(agent_, this));
836 19 : stale_cleaner_->CleanStaleEntries();
837 19 : stale_cleaner_.reset(NULL);
838 19 : return false;
839 : }
840 :
841 0 : void InstanceManager::SetNamespaceStorePath(std::string path) {
842 0 : namespace_store_path_ = path;
843 0 : }
844 :
845 : /*
846 : * InstanceState class
847 : */
848 0 : InstanceState::InstanceState() : DBState(),
849 0 : pid_(0), status_(0), status_type_(0), tasks_running_(0) {
850 0 : }
851 :
852 0 : void InstanceState::Clear() {
853 0 : pid_ = 0;
854 0 : status_ = 0;
855 0 : errors_.clear();
856 0 : cmd_.clear();
857 0 : }
|
