/*
 * Copyright 2018 Google, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License")
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.netflix.spinnaker.clouddriver.kubernetes.v2.caching.agent;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.netflix.spectator.api.Registry;
import com.netflix.spinnaker.cats.agent.AgentDataType;
import com.netflix.spinnaker.cats.agent.CacheResult;
import com.netflix.spinnaker.cats.agent.DefaultCacheResult;
import com.netflix.spinnaker.cats.cache.CacheData;
import com.netflix.spinnaker.cats.provider.ProviderCache;
import com.netflix.spinnaker.clouddriver.kubernetes.KubernetesCloudProvider;
import com.netflix.spinnaker.clouddriver.kubernetes.caching.KubernetesCachingAgent;
import com.netflix.spinnaker.clouddriver.kubernetes.security.KubernetesNamedAccountCredentials;
import com.netflix.spinnaker.clouddriver.kubernetes.v2.security.KubernetesV2Credentials;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.netflix.spinnaker.cats.agent.AgentDataType.Authority.AUTHORITATIVE;
import static com.netflix.spinnaker.clouddriver.kubernetes.v2.caching.Keys.Kind.KUBERNETES_METRIC;

@Slf4j
public class KubernetesMetricCachingAgent extends KubernetesCachingAgent<KubernetesV2Credentials> {
  @Getter
  protected String providerName = KubernetesCloudProvider.getID();

  @Getter
  protected Collection<AgentDataType> providedDataTypes = Collections.unmodifiableCollection(
      Collections.singletonList(AUTHORITATIVE.forType(KUBERNETES_METRIC.toString()))
  );

  protected KubernetesMetricCachingAgent(KubernetesNamedAccountCredentials<KubernetesV2Credentials> namedAccountCredentials,
      ObjectMapper objectMapper,
      Registry registry,
      int agentIndex,
      int agentCount) {
    super(namedAccountCredentials, objectMapper, registry, agentIndex, agentCount);
  }

  @Override
  public CacheResult loadData(ProviderCache providerCache) {
    log.info(getAgentType() + " is starting");
    reloadNamespaces();

    List<CacheData> cacheData = namespaces.stream()
        .map(n -> credentials.topPod(n).stream()
            .map(m -> KubernetesCacheDataConverter.convertPodMetric(accountName, n, m))
        ).flatMap(x -> x)
        .collect(Collectors.toList());

    List<CacheData> invertedRelationships = cacheData.stream()
        .map(KubernetesCacheDataConverter::invertRelationships)
        .flatMap(Collection::stream)
        .collect(Collectors.toList());

    cacheData.addAll(invertedRelationships);

    Map<String, Collection<CacheData>> entries = KubernetesCacheDataConverter.stratifyCacheDataByGroup(KubernetesCacheDataConverter.dedupCacheData(cacheData));
    KubernetesCacheDataConverter.logStratifiedCacheData(getAgentType(), entries);

    return new DefaultCacheResult(entries);
  }
}

@@ -1284,24 +1284,106 @@
 redentials;%0A
+import com.netflix.spinnaker.clouddriver.kubernetes.v2.op.job.KubectlJobExecutor;%0A
 import com.n
@@ -1620,16 +1620,42 @@
 il.Map;%0A
+import java.util.Objects;%0A
 import j
@@ -2786,29 +2786,95 @@
  -%3E 
-credentials.topPod(n)
+%7B%0A              try %7B%0A                return credentials.topPod(n)%0A                    
 .str
@@ -2871,32 +2871,40 @@
       .stream()%0A
+        
             .map
@@ -2974,16 +2974,17 @@
 , n, m))
+;
 %0A       
@@ -2984,17 +2984,436 @@
         
-)
+      %7D catch (KubectlJobExecutor.KubectlException e) %7B%0A                if (e.getMessage().contains(%22not available%22)) %7B%0A                  log.warn(%22Metrics for namespace '%22 + n + %22' in account '%22 + accountName + %22' have not been recorded yet.%22);%0A                  return null;%0A                %7D else %7B%0A                  throw e;%0A                %7D%0A              %7D%0A            %7D%0A        ).filter(Objects::nonNull)%0A        
 .flatMap

address empty namespace metrics (#3109)