run.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988
  1. package cmd
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "os"
  8. "strings"
  9. "time"
  10. "github.com/fatih/color"
  11. api "github.com/porter-dev/porter/api/client"
  12. "github.com/porter-dev/porter/api/types"
  13. "github.com/porter-dev/porter/cli/cmd/utils"
  14. "github.com/spf13/cobra"
  15. batchv1 "k8s.io/api/batch/v1"
  16. v1 "k8s.io/api/core/v1"
  17. rbacv1 "k8s.io/api/rbac/v1"
  18. "k8s.io/apimachinery/pkg/api/resource"
  19. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  20. "k8s.io/apimachinery/pkg/fields"
  21. "k8s.io/apimachinery/pkg/watch"
  22. "k8s.io/kubectl/pkg/util/term"
  23. "k8s.io/apimachinery/pkg/runtime"
  24. "k8s.io/apimachinery/pkg/runtime/schema"
  25. "k8s.io/client-go/kubernetes"
  26. "k8s.io/client-go/rest"
  27. "k8s.io/client-go/tools/clientcmd"
  28. "k8s.io/client-go/tools/remotecommand"
  29. )
  30. var (
  31. namespace string
  32. verbose bool
  33. existingPod bool
  34. nonInteractive bool
  35. containerName string
  36. )
  37. // runCmd represents the "porter run" base command when called
  38. // without any subcommands
  39. var runCmd = &cobra.Command{
  40. Use: "run [release] -- COMMAND [args...]",
  41. Args: cobra.MinimumNArgs(2),
  42. Short: "Runs a command inside a connected cluster container.",
  43. Run: func(cmd *cobra.Command, args []string) {
  44. err := checkLoginAndRun(args, run)
  45. if err != nil {
  46. os.Exit(1)
  47. }
  48. },
  49. }
  50. // cleanupCmd represents the "porter run cleanup" subcommand
  51. var cleanupCmd = &cobra.Command{
  52. Use: "cleanup",
  53. Args: cobra.NoArgs,
  54. Short: "Delete any lingering ephemeral pods that were created with \"porter run\".",
  55. Run: func(cmd *cobra.Command, args []string) {
  56. err := checkLoginAndRun(args, cleanup)
  57. if err != nil {
  58. os.Exit(1)
  59. }
  60. },
  61. }
  62. func init() {
  63. rootCmd.AddCommand(runCmd)
  64. runCmd.PersistentFlags().StringVar(
  65. &namespace,
  66. "namespace",
  67. "default",
  68. "namespace of release to connect to",
  69. )
  70. runCmd.PersistentFlags().BoolVarP(
  71. &existingPod,
  72. "existing_pod",
  73. "e",
  74. false,
  75. "whether to connect to an existing pod",
  76. )
  77. runCmd.PersistentFlags().BoolVarP(
  78. &verbose,
  79. "verbose",
  80. "v",
  81. false,
  82. "whether to print verbose output",
  83. )
  84. runCmd.PersistentFlags().BoolVar(
  85. &nonInteractive,
  86. "non-interactive",
  87. false,
  88. "whether to run in non-interactive mode",
  89. )
  90. runCmd.PersistentFlags().StringVarP(
  91. &containerName,
  92. "container",
  93. "c",
  94. "",
  95. "name of the container inside pod to run the command in",
  96. )
  97. runCmd.AddCommand(cleanupCmd)
  98. }
  99. func run(_ *types.GetAuthenticatedUserResponse, client *api.Client, args []string) error {
  100. execArgs := args[1:]
  101. color.New(color.FgGreen).Println("Running", strings.Join(execArgs, " "), "for release", args[0])
  102. if nonInteractive {
  103. color.New(color.FgBlue).Println("Using non-interactive mode. The first available pod will be used to run the command.")
  104. }
  105. if len(execArgs) > 0 {
  106. release, err := client.GetRelease(
  107. context.Background(), cliConf.Project, cliConf.Cluster, namespace, args[0],
  108. )
  109. if err != nil {
  110. return fmt.Errorf("error fetching release %s: %w", args[0], err)
  111. }
  112. if release.BuildConfig != nil &&
  113. (strings.Contains(release.BuildConfig.Builder, "heroku") ||
  114. strings.Contains(release.BuildConfig.Builder, "paketo")) &&
  115. execArgs[0] != "/cnb/lifecycle/launcher" &&
  116. execArgs[0] != "launcher" {
  117. // this is a buildpacks release using a heroku builder, prepend the launcher
  118. execArgs = append([]string{"/cnb/lifecycle/launcher"}, execArgs...)
  119. }
  120. }
  121. podsSimple, err := getPods(client, namespace, args[0])
  122. if err != nil {
  123. return fmt.Errorf("Could not retrieve list of pods: %s", err.Error())
  124. }
  125. // if length of pods is 0, throw error
  126. var selectedPod podSimple
  127. if len(podsSimple) == 0 {
  128. return fmt.Errorf("At least one pod must exist in this deployment.")
  129. } else if nonInteractive || len(podsSimple) == 1 {
  130. selectedPod = podsSimple[0]
  131. } else {
  132. podNames := make([]string, 0)
  133. for _, podSimple := range podsSimple {
  134. podNames = append(podNames, podSimple.Name)
  135. }
  136. selectedPodName, err := utils.PromptSelect("Select the pod:", podNames)
  137. if err != nil {
  138. return err
  139. }
  140. // find selected pod
  141. for _, podSimple := range podsSimple {
  142. if selectedPodName == podSimple.Name {
  143. selectedPod = podSimple
  144. }
  145. }
  146. }
  147. var selectedContainerName string
  148. if len(selectedPod.ContainerNames) == 0 {
  149. return fmt.Errorf("At least one container must exist in the selected pod.")
  150. } else if len(selectedPod.ContainerNames) == 1 {
  151. if containerName != "" && containerName != selectedPod.ContainerNames[0] {
  152. return fmt.Errorf("provided container %s does not exist in pod %s", containerName, selectedPod.Name)
  153. }
  154. selectedContainerName = selectedPod.ContainerNames[0]
  155. }
  156. if containerName != "" && selectedContainerName == "" {
  157. // check if provided container name exists in the pod
  158. for _, name := range selectedPod.ContainerNames {
  159. if name == containerName {
  160. selectedContainerName = name
  161. break
  162. }
  163. }
  164. if selectedContainerName == "" {
  165. return fmt.Errorf("provided container %s does not exist in pod %s", containerName, selectedPod.Name)
  166. }
  167. }
  168. if selectedContainerName == "" {
  169. if nonInteractive {
  170. return fmt.Errorf("container name must be specified using the --container flag when using non-interactive mode")
  171. }
  172. selectedContainer, err := utils.PromptSelect("Select the container:", selectedPod.ContainerNames)
  173. if err != nil {
  174. return err
  175. }
  176. selectedContainerName = selectedContainer
  177. }
  178. config := &PorterRunSharedConfig{
  179. Client: client,
  180. }
  181. err = config.setSharedConfig()
  182. if err != nil {
  183. return fmt.Errorf("Could not retrieve kube credentials: %s", err.Error())
  184. }
  185. if existingPod {
  186. return executeRun(config, namespace, selectedPod.Name, selectedContainerName, execArgs)
  187. }
  188. return executeRunEphemeral(config, namespace, selectedPod.Name, selectedContainerName, execArgs)
  189. }
  190. func cleanup(_ *types.GetAuthenticatedUserResponse, client *api.Client, _ []string) error {
  191. config := &PorterRunSharedConfig{
  192. Client: client,
  193. }
  194. err := config.setSharedConfig()
  195. if err != nil {
  196. return fmt.Errorf("Could not retrieve kube credentials: %s", err.Error())
  197. }
  198. proceed, err := utils.PromptSelect(
  199. fmt.Sprintf("You have chosen the '%s' namespace for cleanup. Do you want to proceed?", namespace),
  200. []string{"Yes", "No", "All namespaces"},
  201. )
  202. if err != nil {
  203. return err
  204. }
  205. if proceed == "No" {
  206. return nil
  207. }
  208. var podNames []string
  209. color.New(color.FgGreen).Println("Fetching ephemeral pods for cleanup")
  210. if proceed == "All namespaces" {
  211. namespaces, err := config.Clientset.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
  212. if err != nil {
  213. return err
  214. }
  215. for _, namespace := range namespaces.Items {
  216. if pods, err := getEphemeralPods(namespace.Name, config.Clientset); err == nil {
  217. podNames = append(podNames, pods...)
  218. } else {
  219. return err
  220. }
  221. }
  222. } else {
  223. if pods, err := getEphemeralPods(namespace, config.Clientset); err == nil {
  224. podNames = append(podNames, pods...)
  225. } else {
  226. return err
  227. }
  228. }
  229. if len(podNames) == 0 {
  230. color.New(color.FgBlue).Println("No ephemeral pods to delete")
  231. return nil
  232. }
  233. selectedPods, err := utils.PromptMultiselect("Select ephemeral pods to delete", podNames)
  234. if err != nil {
  235. return err
  236. }
  237. for _, podName := range selectedPods {
  238. color.New(color.FgBlue).Printf("Deleting ephemeral pod: %s\n", podName)
  239. err = config.Clientset.CoreV1().Pods(namespace).Delete(
  240. context.Background(), podName, metav1.DeleteOptions{},
  241. )
  242. if err != nil {
  243. return err
  244. }
  245. }
  246. return nil
  247. }
  248. func getEphemeralPods(namespace string, clientset *kubernetes.Clientset) ([]string, error) {
  249. var podNames []string
  250. pods, err := clientset.CoreV1().Pods(namespace).List(
  251. context.Background(), metav1.ListOptions{LabelSelector: "porter/ephemeral-pod"},
  252. )
  253. if err != nil {
  254. return nil, err
  255. }
  256. for _, pod := range pods.Items {
  257. podNames = append(podNames, pod.Name)
  258. }
  259. return podNames, nil
  260. }
  261. type PorterRunSharedConfig struct {
  262. Client *api.Client
  263. RestConf *rest.Config
  264. Clientset *kubernetes.Clientset
  265. RestClient *rest.RESTClient
  266. }
  267. func (p *PorterRunSharedConfig) setSharedConfig() error {
  268. pID := cliConf.Project
  269. cID := cliConf.Cluster
  270. kubeResp, err := p.Client.GetKubeconfig(context.Background(), pID, cID, cliConf.Kubeconfig)
  271. if err != nil {
  272. return err
  273. }
  274. kubeBytes := kubeResp.Kubeconfig
  275. cmdConf, err := clientcmd.NewClientConfigFromBytes(kubeBytes)
  276. if err != nil {
  277. return err
  278. }
  279. restConf, err := cmdConf.ClientConfig()
  280. if err != nil {
  281. return err
  282. }
  283. restConf.GroupVersion = &schema.GroupVersion{
  284. Group: "api",
  285. Version: "v1",
  286. }
  287. restConf.NegotiatedSerializer = runtime.NewSimpleNegotiatedSerializer(runtime.SerializerInfo{})
  288. p.RestConf = restConf
  289. clientset, err := kubernetes.NewForConfig(restConf)
  290. if err != nil {
  291. return err
  292. }
  293. p.Clientset = clientset
  294. restClient, err := rest.RESTClientFor(restConf)
  295. if err != nil {
  296. return err
  297. }
  298. p.RestClient = restClient
  299. return nil
  300. }
  301. type podSimple struct {
  302. Name string
  303. ContainerNames []string
  304. }
  305. func getPods(client *api.Client, namespace, releaseName string) ([]podSimple, error) {
  306. pID := cliConf.Project
  307. cID := cliConf.Cluster
  308. resp, err := client.GetK8sAllPods(context.TODO(), pID, cID, namespace, releaseName)
  309. if err != nil {
  310. return nil, err
  311. }
  312. pods := *resp
  313. res := make([]podSimple, 0)
  314. for _, pod := range pods {
  315. if pod.Status.Phase == v1.PodRunning {
  316. containerNames := make([]string, 0)
  317. for _, container := range pod.Spec.Containers {
  318. containerNames = append(containerNames, container.Name)
  319. }
  320. res = append(res, podSimple{
  321. Name: pod.ObjectMeta.Name,
  322. ContainerNames: containerNames,
  323. })
  324. }
  325. }
  326. return res, nil
  327. }
  328. func executeRun(config *PorterRunSharedConfig, namespace, name, container string, args []string) error {
  329. req := config.RestClient.Post().
  330. Resource("pods").
  331. Name(name).
  332. Namespace(namespace).
  333. SubResource("exec")
  334. for _, arg := range args {
  335. req.Param("command", arg)
  336. }
  337. req.Param("stdin", "true")
  338. req.Param("stdout", "true")
  339. req.Param("tty", "true")
  340. req.Param("container", container)
  341. t := term.TTY{
  342. In: os.Stdin,
  343. Out: os.Stdout,
  344. Raw: true,
  345. }
  346. size := t.GetSize()
  347. sizeQueue := t.MonitorSize(size)
  348. return t.Safe(func() error {
  349. exec, err := remotecommand.NewSPDYExecutor(config.RestConf, "POST", req.URL())
  350. if err != nil {
  351. return err
  352. }
  353. return exec.Stream(remotecommand.StreamOptions{
  354. Stdin: os.Stdin,
  355. Stdout: os.Stdout,
  356. Stderr: os.Stderr,
  357. Tty: true,
  358. TerminalSizeQueue: sizeQueue,
  359. })
  360. })
  361. }
  362. func executeRunEphemeral(config *PorterRunSharedConfig, namespace, name, container string, args []string) error {
  363. existing, err := getExistingPod(config, name, namespace)
  364. if err != nil {
  365. return err
  366. }
  367. newPod, err := createEphemeralPodFromExisting(config, existing, container, args)
  368. if err != nil {
  369. return err
  370. }
  371. podName := newPod.ObjectMeta.Name
  372. // delete the ephemeral pod no matter what
  373. defer deletePod(config, podName, namespace)
  374. color.New(color.FgYellow).Printf("Waiting for pod %s to be ready...", podName)
  375. if err = waitForPod(config, newPod); err != nil {
  376. color.New(color.FgRed).Println("failed")
  377. return handlePodAttachError(err, config, namespace, podName, container)
  378. }
  379. err = checkForPodDeletionCronJob(config)
  380. if err != nil {
  381. return err
  382. }
  383. // refresh pod info for latest status
  384. newPod, err = config.Clientset.CoreV1().
  385. Pods(newPod.Namespace).
  386. Get(context.Background(), newPod.Name, metav1.GetOptions{})
  387. // pod exited while we were waiting. maybe an error maybe not.
  388. // we dont know if the user wanted an interactive shell or not.
  389. // if it was an error the logs hopefully say so.
  390. if isPodExited(newPod) {
  391. color.New(color.FgGreen).Println("complete!")
  392. var writtenBytes int64
  393. writtenBytes, _ = pipePodLogsToStdout(config, namespace, podName, container, false)
  394. if verbose || writtenBytes == 0 {
  395. color.New(color.FgYellow).Println("Could not get logs. Pod events:")
  396. pipeEventsToStdout(config, namespace, podName, container, false)
  397. }
  398. return nil
  399. }
  400. color.New(color.FgGreen).Println("ready!")
  401. color.New(color.FgYellow).Println("Attempting connection to the container. If you don't see a command prompt, try pressing enter.")
  402. req := config.RestClient.Post().
  403. Resource("pods").
  404. Name(podName).
  405. Namespace(namespace).
  406. SubResource("attach")
  407. req.Param("stdin", "true")
  408. req.Param("stdout", "true")
  409. req.Param("tty", "true")
  410. req.Param("container", container)
  411. t := term.TTY{
  412. In: os.Stdin,
  413. Out: os.Stdout,
  414. Raw: true,
  415. }
  416. size := t.GetSize()
  417. sizeQueue := t.MonitorSize(size)
  418. if err = t.Safe(func() error {
  419. exec, err := remotecommand.NewSPDYExecutor(config.RestConf, "POST", req.URL())
  420. if err != nil {
  421. return err
  422. }
  423. return exec.Stream(remotecommand.StreamOptions{
  424. Stdin: os.Stdin,
  425. Stdout: os.Stdout,
  426. Stderr: os.Stderr,
  427. Tty: true,
  428. TerminalSizeQueue: sizeQueue,
  429. })
  430. }); err != nil {
  431. // ugly way to catch no TTY errors, such as when running command "echo \"hello\""
  432. return handlePodAttachError(err, config, namespace, podName, container)
  433. }
  434. if verbose {
  435. color.New(color.FgYellow).Println("Pod events:")
  436. pipeEventsToStdout(config, namespace, podName, container, false)
  437. }
  438. return err
  439. }
  440. func checkForPodDeletionCronJob(config *PorterRunSharedConfig) error {
  441. // try and create the cron job and all of the other required resources as necessary,
  442. // starting with the service account, then role and then a role binding
  443. err := checkForServiceAccount(config)
  444. if err != nil {
  445. return err
  446. }
  447. err = checkForClusterRole(config)
  448. if err != nil {
  449. return err
  450. }
  451. err = checkForRoleBinding(config)
  452. if err != nil {
  453. return err
  454. }
  455. namespaces, err := config.Clientset.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
  456. if err != nil {
  457. return err
  458. }
  459. for _, namespace := range namespaces.Items {
  460. cronJobs, err := config.Clientset.BatchV1().CronJobs(namespace.Name).List(
  461. context.Background(), metav1.ListOptions{},
  462. )
  463. if err != nil {
  464. return err
  465. }
  466. if namespace.Name == "default" {
  467. for _, cronJob := range cronJobs.Items {
  468. if cronJob.Name == "porter-ephemeral-pod-deletion-cronjob" {
  469. return nil
  470. }
  471. }
  472. } else {
  473. for _, cronJob := range cronJobs.Items {
  474. if cronJob.Name == "porter-ephemeral-pod-deletion-cronjob" {
  475. err = config.Clientset.BatchV1().CronJobs(namespace.Name).Delete(
  476. context.Background(), cronJob.Name, metav1.DeleteOptions{},
  477. )
  478. if err != nil {
  479. return err
  480. }
  481. }
  482. }
  483. }
  484. }
  485. // create the cronjob
  486. cronJob := &batchv1.CronJob{
  487. ObjectMeta: metav1.ObjectMeta{
  488. Name: "porter-ephemeral-pod-deletion-cronjob",
  489. },
  490. Spec: batchv1.CronJobSpec{
  491. Schedule: "0 * * * *",
  492. JobTemplate: batchv1.JobTemplateSpec{
  493. Spec: batchv1.JobSpec{
  494. Template: v1.PodTemplateSpec{
  495. Spec: v1.PodSpec{
  496. ServiceAccountName: "porter-ephemeral-pod-deletion-service-account",
  497. RestartPolicy: v1.RestartPolicyNever,
  498. Containers: []v1.Container{
  499. {
  500. Name: "ephemeral-pods-manager",
  501. Image: "public.ecr.aws/o1j4x7p4/porter-ephemeral-pods-manager:latest",
  502. ImagePullPolicy: v1.PullAlways,
  503. Args: []string{"delete"},
  504. },
  505. },
  506. },
  507. },
  508. },
  509. },
  510. },
  511. }
  512. _, err = config.Clientset.BatchV1().CronJobs("default").Create(
  513. context.Background(), cronJob, metav1.CreateOptions{},
  514. )
  515. if err != nil {
  516. return err
  517. }
  518. return nil
  519. }
  520. func checkForServiceAccount(config *PorterRunSharedConfig) error {
  521. namespaces, err := config.Clientset.CoreV1().Namespaces().List(context.Background(), metav1.ListOptions{})
  522. if err != nil {
  523. return err
  524. }
  525. for _, namespace := range namespaces.Items {
  526. serviceAccounts, err := config.Clientset.CoreV1().ServiceAccounts(namespace.Name).List(
  527. context.Background(), metav1.ListOptions{},
  528. )
  529. if err != nil {
  530. return err
  531. }
  532. if namespace.Name == "default" {
  533. for _, svcAccount := range serviceAccounts.Items {
  534. if svcAccount.Name == "porter-ephemeral-pod-deletion-service-account" {
  535. return nil
  536. }
  537. }
  538. } else {
  539. for _, svcAccount := range serviceAccounts.Items {
  540. if svcAccount.Name == "porter-ephemeral-pod-deletion-service-account" {
  541. err = config.Clientset.CoreV1().ServiceAccounts(namespace.Name).Delete(
  542. context.Background(), svcAccount.Name, metav1.DeleteOptions{},
  543. )
  544. if err != nil {
  545. return err
  546. }
  547. }
  548. }
  549. }
  550. }
  551. serviceAccount := &v1.ServiceAccount{
  552. ObjectMeta: metav1.ObjectMeta{
  553. Name: "porter-ephemeral-pod-deletion-service-account",
  554. },
  555. }
  556. _, err = config.Clientset.CoreV1().ServiceAccounts("default").Create(
  557. context.Background(), serviceAccount, metav1.CreateOptions{},
  558. )
  559. if err != nil {
  560. return err
  561. }
  562. return nil
  563. }
  564. func checkForClusterRole(config *PorterRunSharedConfig) error {
  565. roles, err := config.Clientset.RbacV1().ClusterRoles().List(
  566. context.Background(), metav1.ListOptions{},
  567. )
  568. if err != nil {
  569. return err
  570. }
  571. for _, role := range roles.Items {
  572. if role.Name == "porter-ephemeral-pod-deletion-cluster-role" {
  573. return nil
  574. }
  575. }
  576. role := &rbacv1.ClusterRole{
  577. ObjectMeta: metav1.ObjectMeta{
  578. Name: "porter-ephemeral-pod-deletion-cluster-role",
  579. },
  580. Rules: []rbacv1.PolicyRule{
  581. {
  582. APIGroups: []string{""},
  583. Resources: []string{"pods"},
  584. Verbs: []string{"list", "delete"},
  585. },
  586. {
  587. APIGroups: []string{""},
  588. Resources: []string{"namespaces"},
  589. Verbs: []string{"list"},
  590. },
  591. },
  592. }
  593. _, err = config.Clientset.RbacV1().ClusterRoles().Create(
  594. context.Background(), role, metav1.CreateOptions{},
  595. )
  596. if err != nil {
  597. return err
  598. }
  599. return nil
  600. }
  601. func checkForRoleBinding(config *PorterRunSharedConfig) error {
  602. bindings, err := config.Clientset.RbacV1().ClusterRoleBindings().List(
  603. context.Background(), metav1.ListOptions{},
  604. )
  605. if err != nil {
  606. return err
  607. }
  608. for _, binding := range bindings.Items {
  609. if binding.Name == "porter-ephemeral-pod-deletion-cluster-rolebinding" {
  610. return nil
  611. }
  612. }
  613. binding := &rbacv1.ClusterRoleBinding{
  614. ObjectMeta: metav1.ObjectMeta{
  615. Name: "porter-ephemeral-pod-deletion-cluster-rolebinding",
  616. },
  617. RoleRef: rbacv1.RoleRef{
  618. APIGroup: "rbac.authorization.k8s.io",
  619. Kind: "ClusterRole",
  620. Name: "porter-ephemeral-pod-deletion-cluster-role",
  621. },
  622. Subjects: []rbacv1.Subject{
  623. {
  624. APIGroup: "",
  625. Kind: "ServiceAccount",
  626. Name: "porter-ephemeral-pod-deletion-service-account",
  627. Namespace: "default",
  628. },
  629. },
  630. }
  631. _, err = config.Clientset.RbacV1().ClusterRoleBindings().Create(
  632. context.Background(), binding, metav1.CreateOptions{},
  633. )
  634. if err != nil {
  635. return err
  636. }
  637. return nil
  638. }
  639. func waitForPod(config *PorterRunSharedConfig, pod *v1.Pod) error {
  640. var (
  641. w watch.Interface
  642. err error
  643. ok bool
  644. )
  645. // immediately after creating a pod, the API may return a 404. heuristically 1
  646. // second seems to be plenty.
  647. watchRetries := 3
  648. for i := 0; i < watchRetries; i++ {
  649. selector := fields.OneTermEqualSelector("metadata.name", pod.Name).String()
  650. w, err = config.Clientset.CoreV1().
  651. Pods(pod.Namespace).
  652. Watch(context.Background(), metav1.ListOptions{FieldSelector: selector})
  653. if err == nil {
  654. break
  655. }
  656. time.Sleep(time.Second)
  657. }
  658. if err != nil {
  659. return err
  660. }
  661. defer w.Stop()
  662. for {
  663. select {
  664. case <-time.Tick(time.Second):
  665. // poll every second in case we already missed the ready event while
  666. // creating the listener.
  667. pod, err = config.Clientset.CoreV1().
  668. Pods(pod.Namespace).
  669. Get(context.Background(), pod.Name, metav1.GetOptions{})
  670. if isPodReady(pod) || isPodExited(pod) {
  671. return nil
  672. }
  673. case evt := <-w.ResultChan():
  674. pod, ok = evt.Object.(*v1.Pod)
  675. if !ok {
  676. return fmt.Errorf("unexpected object type: %T", evt.Object)
  677. }
  678. if isPodReady(pod) || isPodExited(pod) {
  679. return nil
  680. }
  681. case <-time.After(time.Second * 10):
  682. return errors.New("timed out waiting for pod")
  683. }
  684. }
  685. }
  686. func isPodReady(pod *v1.Pod) bool {
  687. ready := false
  688. conditions := pod.Status.Conditions
  689. for i := range conditions {
  690. if conditions[i].Type == v1.PodReady {
  691. ready = pod.Status.Conditions[i].Status == v1.ConditionTrue
  692. }
  693. }
  694. return ready
  695. }
  696. func isPodExited(pod *v1.Pod) bool {
  697. return pod.Status.Phase == v1.PodSucceeded || pod.Status.Phase == v1.PodFailed
  698. }
  699. func handlePodAttachError(err error, config *PorterRunSharedConfig, namespace, podName, container string) error {
  700. if verbose {
  701. color.New(color.FgYellow).Fprintf(os.Stderr, "Error: %s\n", err)
  702. }
  703. color.New(color.FgYellow).Fprintln(os.Stderr, "Could not open a shell to this container. Container logs:")
  704. var writtenBytes int64
  705. writtenBytes, _ = pipePodLogsToStdout(config, namespace, podName, container, false)
  706. if verbose || writtenBytes == 0 {
  707. color.New(color.FgYellow).Fprintln(os.Stderr, "Could not get logs. Pod events:")
  708. pipeEventsToStdout(config, namespace, podName, container, false)
  709. }
  710. return err
  711. }
  712. func pipePodLogsToStdout(config *PorterRunSharedConfig, namespace, name, container string, follow bool) (int64, error) {
  713. podLogOpts := v1.PodLogOptions{
  714. Container: container,
  715. Follow: follow,
  716. }
  717. req := config.Clientset.CoreV1().Pods(namespace).GetLogs(name, &podLogOpts)
  718. podLogs, err := req.Stream(
  719. context.Background(),
  720. )
  721. if err != nil {
  722. return 0, err
  723. }
  724. defer podLogs.Close()
  725. return io.Copy(os.Stdout, podLogs)
  726. }
  727. func pipeEventsToStdout(config *PorterRunSharedConfig, namespace, name, container string, follow bool) error {
  728. // update the config in case the operation has taken longer than token expiry time
  729. config.setSharedConfig()
  730. // creates the clientset
  731. resp, err := config.Clientset.CoreV1().Events(namespace).List(
  732. context.TODO(),
  733. metav1.ListOptions{
  734. FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.namespace=%s", name, namespace),
  735. },
  736. )
  737. if err != nil {
  738. return err
  739. }
  740. for _, event := range resp.Items {
  741. color.New(color.FgRed).Println(event.Message)
  742. }
  743. return nil
  744. }
  745. func getExistingPod(config *PorterRunSharedConfig, name, namespace string) (*v1.Pod, error) {
  746. return config.Clientset.CoreV1().Pods(namespace).Get(
  747. context.Background(),
  748. name,
  749. metav1.GetOptions{},
  750. )
  751. }
  752. func deletePod(config *PorterRunSharedConfig, name, namespace string) error {
  753. // update the config in case the operation has taken longer than token expiry time
  754. config.setSharedConfig()
  755. err := config.Clientset.CoreV1().Pods(namespace).Delete(
  756. context.Background(),
  757. name,
  758. metav1.DeleteOptions{},
  759. )
  760. if err != nil {
  761. color.New(color.FgRed).Fprintf(os.Stderr, "Could not delete ephemeral pod: %s\n", err.Error())
  762. return err
  763. }
  764. color.New(color.FgGreen).Println("Sucessfully deleted ephemeral pod")
  765. return nil
  766. }
  767. func createEphemeralPodFromExisting(
  768. config *PorterRunSharedConfig,
  769. existing *v1.Pod,
  770. container string,
  771. args []string,
  772. ) (*v1.Pod, error) {
  773. newPod := existing.DeepCopy()
  774. // only copy the pod spec, overwrite metadata
  775. newPod.ObjectMeta = metav1.ObjectMeta{
  776. Name: strings.ToLower(fmt.Sprintf("%s-copy-%s", existing.ObjectMeta.Name, utils.String(4))),
  777. Namespace: existing.ObjectMeta.Namespace,
  778. }
  779. newPod.Status = v1.PodStatus{}
  780. // set restart policy to never
  781. newPod.Spec.RestartPolicy = v1.RestartPolicyNever
  782. // change the command in the pod to the passed in pod command
  783. cmdRoot := args[0]
  784. cmdArgs := make([]string, 0)
  785. // annotate with the ephemeral pod tag
  786. newPod.Labels = make(map[string]string)
  787. newPod.Labels["porter/ephemeral-pod"] = "true"
  788. if len(args) > 1 {
  789. cmdArgs = args[1:]
  790. }
  791. for i := 0; i < len(newPod.Spec.Containers); i++ {
  792. if newPod.Spec.Containers[i].Name == container {
  793. newPod.Spec.Containers[i].Command = []string{cmdRoot}
  794. newPod.Spec.Containers[i].Args = cmdArgs
  795. newPod.Spec.Containers[i].TTY = true
  796. newPod.Spec.Containers[i].Stdin = true
  797. newPod.Spec.Containers[i].StdinOnce = true
  798. if newPod.Spec.Containers[i].Resources.Requests.Cpu() != nil && newPod.Spec.Containers[i].Resources.Requests.Cpu().MilliValue() > 500 {
  799. newPod.Spec.Containers[i].Resources.Limits[v1.ResourceCPU] = resource.MustParse("500m")
  800. newPod.Spec.Containers[i].Resources.Requests[v1.ResourceCPU] = resource.MustParse("500m")
  801. for j := 0; j < len(newPod.Spec.Containers[i].Env); j++ {
  802. if newPod.Spec.Containers[i].Env[j].Name == "PORTER_RESOURCES_CPU" {
  803. newPod.Spec.Containers[i].Env[j].Value = "500m"
  804. break
  805. }
  806. }
  807. }
  808. if newPod.Spec.Containers[i].Resources.Requests.Memory() != nil && newPod.Spec.Containers[i].Resources.Requests.Memory().Value() > 1000*1024*1024 {
  809. newPod.Spec.Containers[i].Resources.Limits[v1.ResourceMemory] = resource.MustParse("1000Mi")
  810. newPod.Spec.Containers[i].Resources.Requests[v1.ResourceMemory] = resource.MustParse("1000Mi")
  811. for j := 0; j < len(newPod.Spec.Containers[i].Env); j++ {
  812. if newPod.Spec.Containers[i].Env[j].Name == "PORTER_RESOURCES_RAM" {
  813. newPod.Spec.Containers[i].Env[j].Value = "1000Mi"
  814. break
  815. }
  816. }
  817. }
  818. }
  819. // remove health checks and probes
  820. newPod.Spec.Containers[i].LivenessProbe = nil
  821. newPod.Spec.Containers[i].ReadinessProbe = nil
  822. newPod.Spec.Containers[i].StartupProbe = nil
  823. }
  824. newPod.Spec.NodeName = ""
  825. // create the pod and return it
  826. return config.Clientset.CoreV1().Pods(existing.ObjectMeta.Namespace).Create(
  827. context.Background(),
  828. newPod,
  829. metav1.CreateOptions{},
  830. )
  831. }