<commit_before>'use strict'

const _ = require('lodash')
const Kube = require('kubernetes-client')
const EventSource = require('./source')
const debug = require('debug')('kubernetes-stream:api')

/** @var config - cached kubeconfig */
let config

function getResourceVersion (object) {
  return _.get(object, 'metadata.resourceVersion')
}

/**
 * A simple helper function to get a request object on
 * the Kube client.
 *
 * @param kind
 * @param apiVersion
 * @param namespace
 * @returns {ApiGroup}
 */
function createClient ({
  kind = 'pods',
  apiVersion = 'v1',
  namespace = undefined
} = {}) {
  let client = new Kube
    .Api(findKubernetesConfig())
    .group(apiVersion)[kind]

  if (namespace !== undefined) {
    client = client.namespace(namespace)
  }

  return client
}

/**
 * Creates an EventSource for the Kubernetes client.
 *
 * @param client
 * @returns {EventSource}
 */
function createEventSource (client = createClient()) {
  return new EventSource(
    /* list  */ (options, callback) => client.get({qs: options}, callback),
    /* watch */ (options, callback) => streamCallback(
      client.get({
        qs: Object.assign({
          watch: true
        }, options)
      }),

      callback
    )
  )
}

/**
 * Find a Kubeconfig from the environment. Looks for
 * KUBECONFIG or KUBERNETES_SERVICE_HOST, or else
 * falls back to reading from ~/.kube/config.
 *
 * @returns {config} suitable for passing to kubernetes-client
 */
function findKubernetesConfig () {
  if (config) {
    return config
  }

  if (process.env.KUBECONFIG) {
    debug('using KUBECONFIG %j', process.env.KUBECONFIG)
    const kubeconfig = Kube.config.loadKubeconfig(process.env.KUBECONFIG)
    config = Kube.config.fromKubeconfig(kubeconfig)
    return config
  }

  if (process.env.KUBERNETES_SERVICE_HOST) {
    debug('using in-cluster api at %s:%d',
      process.env.KUBERNETES_SERVICE_HOST,
      process.env.KUBERNETES_SERVICE_PORT)
    config = Kube.config.getInCluster()
    return config
  }

  debug('falling back to user config')
  config = Kube.config.fromKubeconfig()
  return config
}

/**
 * Take a Readable stream and pipe the chunks into a callback, or
 * null at the end of the stream.
 *
 * Returns a "stop" function that will abort the stream.
 *<commit_msg>experimental oidc support<commit_after>'use strict'

const _ = require('lodash')
const Kube = require('kubernetes-client')
const EventSource = require('./source')
const debug = require('debug')('kubernetes-stream:api')

/** @var config - cached kubeconfig */
let config

function getResourceVersion (object) {
  return _.get(object, 'metadata.resourceVersion')
}

/**
 * A simple helper function to get a request object on
 * the Kube client.
 *
 * @param kind
 * @param apiVersion
 * @param namespace
 * @returns {ApiGroup}
 */
function createClient ({
  kind = 'pods',
  apiVersion = 'v1',
  namespace = undefined
} = {}) {
  let client = new Kube
    .Api(findKubernetesConfig())
    .group(apiVersion)[kind]

  if (namespace !== undefined) {
    client = client.namespace(namespace)
  }

  return client
}

/**
 * Creates an EventSource for the Kubernetes client.
 *
 * @param client
 * @returns {EventSource}
 */
function createEventSource (client = createClient()) {
  return new EventSource(
    /* list  */ (options, callback) => client.get({qs: options}, callback),
    /* watch */ (options, callback) => streamCallback(
      client.get({
        qs: Object.assign({
          watch: true
        }, options)
      }),

      callback
    )
  )
}

/**
 * Find a Kubeconfig from the environment. Looks for
 * KUBECONFIG or KUBERNETES_SERVICE_HOST, or else
 * falls back to reading from ~/.kube/config.
 *
 * @returns {config} suitable for passing to kubernetes-client
 */
function findKubernetesConfig () {
  if (config) {
    return config
  }

  if (process.env.KUBECONFIG) {
    debug('using KUBECONFIG %j', process.env.KUBECONFIG)
    config = parseOidcKubeconfig(process.env.KUBECONFIG)
  }

  else if (process.env.KUBERNETES_SERVICE_HOST) {
    debug('using in-cluster api at %s:%d',
      process.env.KUBERNETES_SERVICE_HOST,
      process.env.KUBERNETES_SERVICE_PORT)
    config = Kube.config.getInCluster()
  }

  else {
    debug('falling back to default kubeconfig')
    config = parseOidcKubeconfig()
  }

  return config
}

function parseOidcKubeconfig (file) {
  const kubeconfig = Kube.config.loadKubeconfig(file)
  const config = Kube.config.fromKubeconfig(kubeconfig)

  if (config && !config.auth) {
    const currentContext = kubeconfig['current-context']
    const context = kubeconfig.contexts
      .find(context => context.name === currentContext).context
    const userConfig = kubeconfig.users
      .find(user => user.name === context.user)

    // hack in id-token support
    const bearer = _.get(userConfig, 'user.auth-provider.config.id-token')
    if (bearer) config.auth = { bearer }
  }

  debug('parsed kubeconfig %j', config)
  return config
}

/**
 * Take a Readable stream and pipe the chunks into a callback, or
 * null at the end of the stream.
 *
 * Returns a "stop" function that will abort the stream.
 *
{'fraction_non_alphanumeric': 0.12225034066575823, 'fraction_numerical': 0.0003893322951138797, 'mean_word_length': 3.3989726027397262, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4965343', 'n_tokens_mistral': 1608, 'n_tokens_neox': 1601, 'n_words': 436}