source.dussan.org Git - sonarqube.git/blob

   1 /*
   2  * SonarQube
   3  * Copyright (C) 2009-2017 SonarSource SA
   4  * mailto:info AT sonarsource DOT com
   5  *
   6  * This program is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 3 of the License, or (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General Public License
  17  * along with this program; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  19  */
  20 package org.sonar.server.computation.taskprocessor;
  21
  22 import com.google.common.base.Optional;
  23 import javax.annotation.Nullable;
  24 import org.sonar.api.utils.log.Logger;
  25 import org.sonar.api.utils.log.Loggers;
  26 import org.sonar.ce.log.CeLogging;
  27 import org.sonar.ce.queue.CeTask;
  28 import org.sonar.ce.queue.CeTaskResult;
  29 import org.sonar.ce.taskprocessor.CeTaskProcessor;
  30 import org.sonar.core.util.logs.Profiler;
  31 import org.sonar.db.ce.CeActivityDto;
  32 import org.sonar.server.computation.queue.InternalCeQueue;
  33
  34 import static java.lang.String.format;
  35
  36 public class CeWorkerCallableImpl implements CeWorkerCallable {
  37
  38   private static final Logger LOG = Loggers.get(CeWorkerCallableImpl.class);
  39
  40   private final InternalCeQueue queue;
  41   private final CeLogging ceLogging;
  42   private final CeTaskProcessorRepository taskProcessorRepository;
  43
  44   public CeWorkerCallableImpl(InternalCeQueue queue, CeLogging ceLogging, CeTaskProcessorRepository taskProcessorRepository) {
  45     this.queue = queue;
  46     this.ceLogging = ceLogging;
  47     this.taskProcessorRepository = taskProcessorRepository;
  48   }
  49
  50   @Override
  51   public Boolean call() throws Exception {
  52     Optional<CeTask> ceTask = tryAndFindTaskToExecute();
  53     if (!ceTask.isPresent()) {
  54       return false;
  55     }
  56
  57     try {
  58       executeTask(ceTask.get());
  59     } catch (Exception e) {
  60       LOG.error(format("An error occurred while executing task with uuid '%s'", ceTask.get().getUuid()), e);
  61     }
  62     return true;
  63   }
  64
  65   private Optional<CeTask> tryAndFindTaskToExecute() {
  66     try {
  67       return queue.peek();
  68     } catch (Exception e) {
  69       LOG.error("Failed to pop the queue of analysis reports", e);
  70     }
  71     return Optional.absent();
  72   }
  73
  74   private void executeTask(CeTask task) {
  75     ceLogging.initForTask(task);
  76     Profiler ceProfiler = startActivityProfiler(task);
  77
  78     CeActivityDto.Status status = CeActivityDto.Status.FAILED;
  79     CeTaskResult taskResult = null;
  80     Throwable error = null;
  81     try {
  82       // TODO delegate the message to the related task processor, according to task type
  83       Optional<CeTaskProcessor> taskProcessor = taskProcessorRepository.getForCeTask(task);
  84       if (taskProcessor.isPresent()) {
  85         taskResult = taskProcessor.get().process(task);
  86         status = CeActivityDto.Status.SUCCESS;
  87       } else {
  88         LOG.error("No CeTaskProcessor is defined for task of type {}. Plugin configuration may have changed", task.getType());
  89         status = CeActivityDto.Status.FAILED;
  90       }
  91     } catch (Throwable e) {
  92       LOG.error(format("Failed to execute task %s", task.getUuid()), e);
  93       error = e;
  94     } finally {
  95       finalizeTask(task, ceProfiler, status, taskResult, error);
  96     }
  97   }
  98
  99   private void finalizeTask(CeTask task, Profiler ceProfiler, CeActivityDto.Status status,
 100     @Nullable CeTaskResult taskResult, @Nullable Throwable error) {
 101     try {
 102       queue.remove(task, status, taskResult, error);
 103     } catch (Exception e) {
 104       LOG.error(format("Failed to finalize task with uuid '%s' and persist its state to db", task.getUuid()), e);
 105     } finally {
 106       stopActivityProfiler(ceProfiler, task, status);
 107       ceLogging.clearForTask();
 108     }
 109   }
 110
 111   private static Profiler startActivityProfiler(CeTask task) {
 112     Profiler profiler = Profiler.create(LOG);
 113     addContext(profiler, task);
 114     return profiler.startInfo("Execute task");
 115   }
 116
 117   private static void stopActivityProfiler(Profiler profiler, CeTask task, CeActivityDto.Status status) {
 118     addContext(profiler, task);
 119     if (status == CeActivityDto.Status.FAILED) {
 120       profiler.stopError("Executed task");
 121     } else {
 122       profiler.stopInfo("Executed task");
 123     }
 124   }
 125
 126   private static void addContext(Profiler profiler, CeTask task) {
 127     profiler
 128       .logTimeLast(true)
 129       .addContext("project", task.getComponentKey())
 130       .addContext("type", task.getType())
 131       .addContext("id", task.getUuid());
 132     String submitterLogin = task.getSubmitterLogin();
 133     if (submitterLogin != null) {
 134       profiler.addContext("submitter", submitterLogin);
 135     }
 136   }
 137
 138 }