func()

in oracle/pkg/database/dbdaemon/dbdaemon_server.go [523:614]
65 lines of code
11 McCabe index (conditional complexity)

func (s *Server) dataPumpImport(ctx context.Context, req *dbdpb.DataPumpImportRequest) (*dbdpb.DataPumpImportResponse, error) {
	s.syncJobs.pdbLoadMutex.Lock()
	defer s.syncJobs.pdbLoadMutex.Unlock()

	importFilename := "import.dmp"
	importMetaFile := importFilename + ".meta"
	logFilename := "import.log"

	pdbPath := fmt.Sprintf(consts.PDBPathPrefix, consts.DataMount, s.databaseSid.val, strings.ToUpper(req.PdbName))
	dumpDir := filepath.Join(pdbPath, consts.DpdumpDir.Linux)
	klog.InfoS("dbdaemon/dataPumpImport", "dumpDir", dumpDir)

	dmpReader, err := s.gcsUtil.Download(ctx, req.GcsPath)
	if err != nil {
		return nil, fmt.Errorf("dbdaemon/dataPumpImport: initiating GCS download failed: %v", err)
	}
	defer dmpReader.Close()

	importFileFullPath := filepath.Join(dumpDir, importFilename)
	if err := s.osUtil.createFile(importFileFullPath, dmpReader); err != nil {
		return nil, fmt.Errorf("dbdaemon/dataPumpImport: download from GCS failed: %v", err)
	}
	klog.Infof("dbdaemon/dataPumpImport: downloaded import dmp file from %s to %s", req.GcsPath, importFileFullPath)
	defer func() {
		if err := s.osUtil.removeFile(importFileFullPath); err != nil {
			klog.Warning(fmt.Sprintf("dbdaemon/dataPumpImport: failed to remove import dmp file after import: %v", err))
		}
	}()

	impdpTarget, err := security.SetupUserPwConnStringOnServer(ctx, s, consts.PDBLoaderUser, req.PdbName, req.DbDomain)
	if err != nil {
		return nil, fmt.Errorf("dbdaemon/dataPumpImport: failed to alter user %s", consts.PDBLoaderUser)
	}

	// Tested with:
	// impdp \"/ as sysdba\" sqlfile=test_meta_tables.sql dumpfile=prod_export_cmms02072022.dmp directory=REFRESH_DUMP_DIR NOLOGFILE=YES FULL=Y INCLUDE=TABLESPACE INCLUDE=TABLESPACE_QUOTA INCLUDE=USER PARALLEL=4
	// We dont dump tables because this can be slow O(minutes), but it would be more precise if needed it can be added.
	tsCheckParams := []string{impdpTarget}
	tsCheckParams = append(tsCheckParams, filterParamsForMetadata(req.CommandParams)...)
	tsCheckParams = append(tsCheckParams, fmt.Sprintf("directory=%s", consts.DpdumpDir.Oracle))
	tsCheckParams = append(tsCheckParams, "dumpfile="+importFilename)
	tsCheckParams = append(tsCheckParams, "sqlfile="+importMetaFile)
	tsCheckParams = append(tsCheckParams, "nologfile=YES")
	tsCheckParams = append(tsCheckParams, "include=TABLESPACE")
	tsCheckParams = append(tsCheckParams, "include=USER")
	tsCheckParams = append(tsCheckParams, "include=TABLESPACE_QUOTA")

	if err := s.runCommand(impdp(s.databaseHome), tsCheckParams); err != nil {
		// On error code 5 (EX_SUCC_ERR), process completed reached the
		// end but data in the DMP might have been skipped (foreign
		// schemas, already imported tables, even failed schema imports
		// because the DMP didn't include CREATE USER statements.)
		if !s.osUtil.isReturnCodeEqual(err, 5) {
			return nil, fmt.Errorf("data pump import metadata gathering failed, err = %v", err)
		}

		klog.Warning("dbdaemon/dataPumpImport: metadata gathering completed with EX_SUCC_ERR")
	}

	metaFullPath := filepath.Join(dumpDir, importMetaFile)
	s.createTablespacesFromSqlfile(ctx, metaFullPath, req.PdbName)

	params := []string{impdpTarget}
	params = append(params, req.CommandParams...)
	params = append(params, fmt.Sprintf("directory=%s", consts.DpdumpDir.Oracle))
	params = append(params, "dumpfile="+importFilename)
	params = append(params, "logfile="+logFilename)

	if err := s.runCommand(impdp(s.databaseHome), params); err != nil {
		// On error code 5 (EX_SUCC_ERR), process completed reached the
		// end but data in the DMP might have been skipped (foreign
		// schemas, already imported tables, even failed schema imports
		// because the DMP didn't include CREATE USER statements.)
		if !s.osUtil.isReturnCodeEqual(err, 5) {
			return nil, fmt.Errorf("data pump import failed, err = %v", err)
		}

		klog.Warning("dbdaemon/dataPumpImport: completed with EX_SUCC_ERR")
	}

	if len(req.GcsLogPath) > 0 {
		logFullPath := filepath.Join(dumpDir, logFilename)

		if err := s.gcsUtil.UploadFile(ctx, req.GcsLogPath, logFullPath, contentTypePlainText); err != nil {
			return nil, fmt.Errorf("dbdaemon/dataPumpImport: import completed successfully, failed to upload import log to GCS: %v", err)
		}

		klog.Infof("dbdaemon/dataPumpImport: uploaded import log to %s", req.GcsLogPath)
	}

	return &dbdpb.DataPumpImportResponse{}, nil
}