Getting logs from k8s may cause OutOfMemory error
# Background
For some workloads, container logs may become very large: up to several gigabytes. When we try to fetch such logs it is likely to cause OutOfMemory error, since k8s library will try to load it into a single String object. We need to limit log's size to prevent such problems.
# Action items

- Introduce a new `SystemPreference` with pod log size limit in lines
- Method for getting pod logs shall fetch tail log lines with respect to this limit
- If actual pod logs exceeds limit add a prefix to log text to show that log is truncated

diff --git a/api/src/main/java/com/epam/pipeline/common/MessageConstants.java b/api/src/main/java/com/epam/pipeline/common/MessageConstants.java
index 6f9d97df7..2fa2d54f2 100644
--- a/api/src/main/java/com/epam/pipeline/common/MessageConstants.java
+++ b/api/src/main/java/com/epam/pipeline/common/MessageConstants.java
@@ -158,6 +158,7 @@ public final class MessageConstants {
     public static final String ERROR_KUBE_SERVICE_IP_UNDEFINED = "error.kube.service.ip.undefined";
     public static final String ERROR_KUBE_SERVICE_PORT_UNDEFINED = "error.kube.service.port.undefined";
     public static final String ERROR_NODE_DOWN_TIMEOUT = "error.kube.node.down.timeout";
+    public static final String LOG_WAS_TRUNCATED = "log.truncated";
 
     // Data storage messages
     public static final String ERROR_DATASTORAGE_NOT_FOUND = "error.datastorage.not.found";
diff --git a/api/src/main/java/com/epam/pipeline/manager/cluster/KubernetesManager.java b/api/src/main/java/com/epam/pipeline/manager/cluster/KubernetesManager.java
index a474d1e04..73303d155 100644
--- a/api/src/main/java/com/epam/pipeline/manager/cluster/KubernetesManager.java
+++ b/api/src/main/java/com/epam/pipeline/manager/cluster/KubernetesManager.java
@@ -75,6 +75,7 @@ public class KubernetesManager {
     private static final Logger LOGGER = LoggerFactory.getLogger(KubernetesManager.class);
     private static final String DEFAULT_SVC_SCHEME = "http";
     private static final int NODE_PULL_TIMEOUT = 200;
+    private static final String NEW_LINE = "\n";
 
     private ObjectMapper mapper = new JsonMapper();
 
@@ -177,15 +178,24 @@ public class KubernetesManager {
         }
     }
 
-    public String getPodLogs(String podId) {
-        try (KubernetesClient client = getKubernetesClient()) {
-            return client.pods().inNamespace(kubeNamespace).withName(podId).getLog();
+    public String getPodLogs(final String podId, final int limit) {
+        try (final KubernetesClient client = getKubernetesClient()) {
+            final String tail = client.pods().inNamespace(kubeNamespace)
+                    .withName(podId)
+                    .tailingLines(limit + 1).getLog();
+            return isLogTruncated(tail, limit)
+                    ? messageHelper.getMessage(MessageConstants.LOG_WAS_TRUNCATED) + NEW_LINE + tail
+                    : tail;
         } catch (KubernetesClientException e) {
             LOGGER.error(e.getMessage(), e);
             return null;
         }
     }
 
+    private boolean isLogTruncated(final String tail, final int limit) {
+        return tail.split(NEW_LINE).length > limit;
+    }
+
     /**
      * Updates current status with node statuses that indicate node failure.
      * @param status current status
diff --git a/api/src/main/java/com/epam/pipeline/manager/pipeline/RunLogManager.java b/api/src/main/java/com/epam/pipeline/manager/pipeline/RunLogManager.java
index 99731fad5..3847ee8f1 100644
--- a/api/src/main/java/com/epam/pipeline/manager/pipeline/RunLogManager.java
+++ b/api/src/main/java/com/epam/pipeline/manager/pipeline/RunLogManager.java
@@ -28,6 +28,8 @@ import com.epam.pipeline.entity.pipeline.TaskStatus;
 import com.epam.pipeline.common.MessageConstants;
 import com.epam.pipeline.dao.pipeline.PipelineRunDao;
 import com.epam.pipeline.manager.cluster.KubernetesManager;
+import com.epam.pipeline.manager.preference.PreferenceManager;
+import com.epam.pipeline.manager.preference.SystemPreferences;
 import com.epam.pipeline.utils.LogsFormatter;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -61,6 +63,9 @@ public class RunLogManager {
     @Autowired
     private KubernetesManager kubernetesManager;
 
+    @Autowired
+    private PreferenceManager preferenceManager;
+
     private RunLogManager self;
 
     private LogsFormatter logsFormatter = new LogsFormatter();
@@ -193,8 +198,9 @@ public class RunLogManager {
     }
 
     private List<RunLog> getPodLogs(PipelineRun run) {
+        int logLimit = preferenceManager.getPreference(SystemPreferences.SYSTEM_LIMIT_LOG_LINES);
         String logText = StringUtils.isBlank(run.getPodIP()) ?
-                "Node initialization in progress." : kubernetesManager.getPodLogs(run.getPodId());
+                "Node initialization in progress." : kubernetesManager.getPodLogs(run.getPodId(), logLimit);
         RunLog log = new RunLog();
         log.setRunId(run.getId());
         log.setStatus(TaskStatus.RUNNING);
diff --git a/api/src/main/java/com/epam/pipeline/manager/preference/SystemPreferences.java b/api/src/main/java/com/epam/pipeline/manager/preference/SystemPreferences.java
index 5e657b35a..3d14c8ab4 100644
--- a/api/src/main/java/com/epam/pipeline/manager/preference/SystemPreferences.java
+++ b/api/src/main/java/com/epam/pipeline/manager/preference/SystemPreferences.java
@@ -367,6 +367,13 @@ public class SystemPreferences {
      */
     public static final IntPreference SYSTEM_RESOURCE_MONITORING_PERIOD = new IntPreference(
         "system.resource.monitoring.period", 60000, SYSTEM_GROUP, isGreaterThan(10000));
+
+    /**
+     * Controls the amount of pod logs to be loaded
+     */
+    public static final IntPreference SYSTEM_LIMIT_LOG_LINES = new IntPreference(
+            "system.log.line.limit", 8000, SYSTEM_GROUP, isGreaterThan(0));
+
     /**
      * Level of CPU load, below which a Run is considered `idle`
      */
diff --git a/api/src/main/resources/messages.properties b/api/src/main/resources/messages.properties
index 061caa056..f9239d3fb 100644
--- a/api/src/main/resources/messages.properties
+++ b/api/src/main/resources/messages.properties
@@ -159,6 +159,7 @@ error.kube.master.ip.not.found=Kube master ip was not found.
 error.kube.service.ip.undefined=Kubernetes service {0} external IP is not defined.
 error.kube.service.port.undefined=Kubernetes service {0} port is not defined.
 error.kube.node.down.timeout=Node {0} was not removed in requested period.
+log.truncated=< ... Some of the preceding lines were skipped due to log size limit ...>
 
 # Data sources messages
 error.datastorage.not.found=Error: data storage with id: ''{0}'' was not found.
['api/src/main/java/com/epam/pipeline/manager/preference/SystemPreferences.java', 'api/src/main/resources/messages.properties', 'api/src/main/java/com/epam/pipeline/common/MessageConstants.java', 'api/src/main/java/com/epam/pipeline/manager/cluster/KubernetesManager.java', 'api/src/main/java/com/epam/pipeline/manager/pipeline/RunLogManager.java']