lightning: reduce the "unable to get keyspace name" log level to DEBUG (

pingcap#54243) (pingcap#55053)

close pingcap#54232

lightning/pkg/importer/import.go

@@ -408,6 +408,12 @@ func NewImportControllerWithPauser(
 		}
 		p.TaskType = taskType
 
+		// TODO: we should not need to check config here.
+		// Instead, we should perform the following during switch mode:
+		//  1. for each tikv, try to switch mode without any ranges.
+		//  2. if it returns normally, it means the store is using a raft-v1 engine.
+		//  3. if it returns the `partitioned-raft-kv only support switch mode with range set` error,
+		//     it means the store is a raft-v2 engine and we will include the ranges from now on.
 		isRaftKV2, err := common.IsRaftKV2(ctx, db)
 		if err != nil {
 			log.FromContext(ctx).Warn("check isRaftKV2 failed", zap.Error(err))

lightning/pkg/server/lightning.go

@@ -564,6 +564,16 @@ func (l *Lightning) run(taskCtx context.Context, taskCfg *config.Config, o *opti
 		keyspaceName = taskCfg.TikvImporter.KeyspaceName
 		if keyspaceName == "" {
 			keyspaceName, err = getKeyspaceName(db)
+			if err != nil && common.IsAccessDeniedNeedConfigPrivilegeError(err) {
+				// if the cluster is not multitenant we don't really need to know about the keyspace.
+				// since the doc does not say we require CONFIG privilege,
+				// spelling out the Access Denied error just confuses the users.
+				// hide such allowed errors unless log level is DEBUG.
+				o.logger.Info("keyspace is unspecified and target user has no config privilege, assuming dedicated cluster")
+				if o.logger.Level() > zapcore.DebugLevel {
+					err = nil
+				}
+			}
 			if err != nil {
 				o.logger.Warn("unable to get keyspace name, lightning will use empty keyspace name", zap.Error(err))
 			}

pkg/lightning/common/util.go

@@ -696,3 +696,9 @@ func IsRaftKV2(ctx context.Context, db *sql.DB) (bool, error) {
 	}
 	return false, rows.Err()
 }
+
+// IsAccessDeniedNeedConfigPrivilegeError checks if err is generated from a query to TiDB which failed due to missing CONFIG privilege.
+func IsAccessDeniedNeedConfigPrivilegeError(err error) bool {
+	e, ok := err.(*mysql.MySQLError)
+	return ok && e.Number == errno.ErrSpecificAccessDenied && strings.Contains(e.Message, "CONFIG")
+}