Feat: dqlite dump for core, dpm, integrations and license databases

Author: mrajagopal

pkg/data_collector/data_collector.go

@@ -268,86 +268,6 @@ func (c *DataCollector) AllNamespacesExist() bool {
 	return allExist
 }
 
-// // CopyFileFromPod copies a file from a pod's container to the local filesystem.
-// func (c *DataCollector) CopyFileFromPod(namespace, pod, container, srcPath, destPath string, ctx context.Context) error {
-// 	cmd := []string{"tar", "cf", "-", "-C", filepath.Dir(srcPath), filepath.Base(srcPath)}
-// 	req := c.K8sCoreClientSet.CoreV1().RESTClient().Post().
-// 		Namespace(namespace).
-// 		Resource("pods").
-// 		Name(pod).
-// 		SubResource("exec").
-// 		VersionedParams(&corev1.PodExecOptions{
-// 			Container: container,
-// 			Command:   cmd,
-// 			Stdin:     false,
-// 			Stdout:    true,
-// 			Stderr:    true,
-// 			TTY:       false,
-// 		}, scheme.ParameterCodec)
-
-// 	exec, err := remotecommand.NewSPDYExecutor(c.K8sRestConfig, "POST", req.URL())
-// 	if err != nil {
-// 		return err
-// 	}
-// 	fmt.Printf("Started remote command\n")
-// 	reader, writer := io.Pipe()
-// 	// var wg sync.WaitGroup
-// 	var streamErr error
-// 	// wg.Add(1)
-// 	go func() {
-// 		defer writer.Close()
-// 		// 	defer wg.Done()
-// 		streamErr = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
-// 			Stdout: writer,
-// 			Stderr: os.Stderr,
-// 		})
-// 	}()
-
-// 	tr := tar.NewReader(reader)
-// 	fmt.Printf("New Reader started\n")
-// 	var copyErr error
-// 	for {
-// 		header, err := tr.Next()
-// 		if err == io.EOF {
-// 			fmt.Printf("Reached end of tar stream\n")
-// 			break
-// 		}
-// 		if err != nil {
-// 			copyErr = err
-// 			break
-// 		}
-// 		if header.Typeflag == tar.TypeReg {
-// 			fmt.Printf("Copying file %s to destPath %s\n", header.Name, destPath)
-// 			outFile, err := os.Create(destPath)
-// 			if err != nil {
-// 				copyErr = err
-// 				break
-// 			}
-// 			fmt.Printf("Copying file %s to outFile %s\n", header.Name, outFile.Name())
-// 			defer outFile.Close()
-// 			_, err = io.Copy(outFile, tr)
-
-// 			if err != nil {
-// 				copyErr = err
-// 				break
-// 			}
-// 		}
-// 	}
-// 	// Wait for the goroutine to finish
-// 	fmt.Printf("Waiting for stream to finish\n")
-// 	// wg.Wait()
-// 	fmt.Printf("Stream finished\n")
-// 	if copyErr != nil {
-// 		fmt.Printf("Error copying file: %v\n", copyErr)
-// 		return copyErr
-// 	}
-// 	if streamErr != nil {
-// 		fmt.Printf("Error executing command in pod: %v\n", streamErr)
-// 		return streamErr
-// 	}
-// 	return nil
-// }
-
 // CopyFileFromPod copies a file from a pod's container to the local filesystem.
 func (c *DataCollector) CopyFileFromPod(namespace, pod, container, srcPath, destPath string, ctx context.Context) error {
 	cmd := []string{"tar", "cf", "-", "-C", filepath.Dir(srcPath), filepath.Base(srcPath)}
@@ -377,7 +297,6 @@ func (c *DataCollector) CopyFileFromPod(namespace, pod, container, srcPath, dest
 		Stderr: &stderr,
 	})
 	if err != nil {
-		// return fmt.Errorf("error in streaming: %w. Stderr: %s", err, stderr.String())
 		return err
 	}
 
@@ -392,23 +311,17 @@ func (c *DataCollector) CopyFileFromPod(namespace, pod, container, srcPath, dest
 	// Untar the stream and write the content to the local file
 	tarReader := tar.NewReader(&stdout)
 	for {
-		// fmt.Printf("Reading tar stream\n")
-		// fmt.Println("Tar output length:", stdout.Len())
 		header, err := tarReader.Next()
 
 		if err == io.EOF {
-			// fmt.Printf("Reached end of tar stream\n")
 			break // End of tar archive
 		}
 		if err != nil {
-			// return fmt.Errorf("error reading tar stream: %w", err)
 			return err
 		}
 
 		// Ensure the tar file matches the expected file path
-		// fmt.Printf("Header Name: %s\n", header.Name)
 		if header.Name == filepath.Base(srcPath) {
-			// fmt.Printf("Copying file %s to destPath %s\n", header.Name, destPath)
 			_, err = io.Copy(localFile, tarReader)
 			if err != nil {
 				return fmt.Errorf("failed to write to local file: %w", err)

pkg/jobs/nim_job_list.go

@@ -186,38 +186,59 @@ func NIMJobList() []Job {
 			},
 		},
 		{
-			Name:    "exec-dqlite-dump-core",
+			Name:    "exec-dqlite-dump",
 			Timeout: time.Second * 30,
 			Execute: func(dc *data_collector.DataCollector, ctx context.Context, ch chan JobResult) {
 				jobResult := JobResult{Files: make(map[string][]byte), Error: nil}
-				containerName := "core"
-				dbName := "core"
-				outputFile := "/tmp/core.sql"
-				dbAddr := "0.0.0.0:7891"
+
+				dbConfigs := []struct {
+					dbName        string
+					containerName string
+					outputFile    string
+					dbAddr        string
+				}{
+					{"core", "core", "/tmp/core.sql", "0.0.0.0:7891"},
+					{"dpm", "dpm", "/tmp/dpm.sql", "0.0.0.0:7890"},
+					{"integrations", "integrations", "/tmp/integrations.sql", "0.0.0.0:7892"},
+					{"license", "integrations", "/tmp/license.sql", "0.0.0.0:7893"},
+					// Add more containers as needed
+				}
 
 				// /etc/nms/scripts/dqlite-backup -n core -c /etc/nms/nms.conf -a 0.0.0.0:7891 -o /tmp/core.sql -k
-				command := []string{dqliteBackupPath, "-n", dbName, "-c", nmsConfigPath, "-a", dbAddr, "-o", outputFile, "-k"}
+
 				for _, namespace := range dc.Namespaces {
 					pods, err := dc.K8sCoreClientSet.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{})
 					if err != nil {
 						dc.Logger.Printf("\tCould not retrieve pod list for namespace %s: %v\n", namespace, err)
 					} else {
-						for _, pod := range pods.Items {
-							if strings.Contains(pod.Name, containerName) {
-								res, err := dc.PodExecutor(namespace, pod.Name, containerName, command, ctx)
-								if err != nil {
-									jobResult.Error = err
-									dc.Logger.Printf("\tCommand execution %s failed for pod %s in namespace %s: %v\n", command, pod.Name, namespace, err)
-								} else {
-									jobResult.Files[filepath.Join(dc.BaseDir, "exec", namespace, pod.Name+"__dqlite-dump-"+containerName+".txt")] = res
-
-									// Move the dumped file to the base directory
-									destPathFilename := filepath.Join(dc.BaseDir, "exec", namespace, pod.Name+"__dqlite-dump-"+filepath.Base(outputFile))
-									if err := dc.CopyFileFromPod(namespace, pod.Name, containerName, outputFile, destPathFilename, ctx); err != nil {
+						for _, config := range dbConfigs {
+							command := []string{dqliteBackupPath, "-n", config.dbName, "-c", nmsConfigPath, "-a", config.dbAddr, "-o", config.outputFile, "-k"}
+							for _, pod := range pods.Items {
+								if strings.Contains(pod.Name, config.containerName) {
+									res, err := dc.PodExecutor(namespace, pod.Name, config.containerName, command, ctx)
+									if err != nil {
 										jobResult.Error = err
-										dc.Logger.Printf("\tFailed to copy dumped file for pod %s in namespace %s: %v\n", pod.Name, namespace, err)
+										dc.Logger.Printf("\tCommand execution %s failed for pod %s in namespace %s: %v\n", command, pod.Name, namespace, err)
 									} else {
-										dc.Logger.Printf("\tSuccessfully copied dumped file for pod %s in namespace %s\n", pod.Name, namespace)
+										jobResult.Files[filepath.Join(dc.BaseDir, "exec", namespace, pod.Name+"__dqlite-dump-"+config.dbName+".txt")] = res
+
+										// Copy the dumped file from the pod to the host
+										destPathFilename := filepath.Join(dc.BaseDir, "exec", namespace, pod.Name+"__dqlite-dump-"+filepath.Base(config.outputFile))
+										if err := dc.CopyFileFromPod(namespace, pod.Name, config.containerName, config.outputFile, destPathFilename, ctx); err != nil {
+											jobResult.Error = err
+											dc.Logger.Printf("\tFailed to copy dumped file %s from pod %s in namespace %s to %s: %v\n", config.outputFile, pod.Name, namespace, destPathFilename, err)
+										} else {
+											dc.Logger.Printf("\tSuccessfully copied dumped file %s from pod %s in namespace %s to %s\n", config.outputFile, pod.Name, namespace, destPathFilename)
+										}
+
+										// Remove/delete the dumped file from the pod
+										_, err := dc.PodExecutor(namespace, pod.Name, config.containerName, []string{"rm", "-f", config.outputFile}, ctx)
+										if err != nil {
+											jobResult.Error = err
+											dc.Logger.Printf("\tFailed to remove dumped file %s from pod %s in namespace %s: %v\n", config.outputFile, pod.Name, namespace, err)
+										} else {
+											dc.Logger.Printf("\tSuccessfully removed dumped file %s from pod %s in namespace %s\n", config.outputFile, pod.Name, namespace)
+										}
 									}
 								}
 							}