mk-oracle: fix monitoring of Oracle 12.1 and 12.2

Neither release was monitored at all.
The version was determined by VERSION_FULL, with fallback to
V$VERSION.BANNER_FULL. However, both were unavailable before 18c.

- version probe: VERSION_FULL where it exists, else VERSION. Retried on
  ORA-00904 alone, so a permissions or connection error still surfaces
  instead of reading as an ancient release.
- version extraction no longer takes the last whitespace-separated token,
  which is "Production" in a one-line banner.

Releases below the supported floor of 12.1.0.2 are not accommodated: they
fail on the absent V$DATABASE.CDB column, with the version the probe
established already in the log.

TESTS: Manual and automated. Confirmed against real Oracle DBs in Docker.
Tested DB versions: 12.1 and 12.2 now emit their sections where they emitted
none, 19.3 and 23.26 are unchanged, and 11.2 fails on the absent column.

CMK-37730

Change-Id: Ifb0484f201ca1a93ceb36f04dd9f3e44ce48d82b

Author: andiumbreit

packages/mk-oracle/src/ora_sql/backend.rs

@@ -479,7 +479,11 @@ pub fn obtain_config_credentials(auth: &Authentication) -> Option<Credentials> {
 pub(crate) mod test_support {
     //! Fake [`OraDbEngine`] and helpers for tests.
     use super::*;
-    use crate::ora_sql::sqls::query::internal::{INSTANCE_INFO_SQL_TEXT_NEW, PDB_DISCOVERY_SQL};
+    use crate::ora_sql::sqls::query::internal::{
+        INSTANCE_INFO_SQL_TEXT_NEW, INSTANCE_INFO_SQL_TEXT_OLD, INSTANCE_VERSION,
+        INSTANCE_VERSION_FULL, PDB_DISCOVERY_SQL,
+    };
+    use std::sync::{Arc, Mutex};
 
     /// A `v$instance` row; parsing reads only columns 0 (name), 2 (version), 4 (cdb).
     pub fn instance_row(name: &str, version: &str, cdb: &str) -> Vec<String> {
@@ -498,6 +502,21 @@ pub(crate) mod test_support {
         pub instance_rows: Vec<Vec<String>>,
         pub pdb_rows: Vec<Vec<String>>,
         pub default_rows: Vec<Vec<String>>,
+        /// A query naming any of these is answered with ORA-00904.
+        pub absent_columns: Vec<String>,
+        /// Shared with every clone from `clone_box`, so one run collects here.
+        pub asked: Arc<Mutex<Vec<String>>>,
+        pub version_rows: Vec<Vec<String>>,
+    }
+
+    impl MiniOra {
+        fn missing_column_in(&self, query: &str) -> Option<&str> {
+            let upper = query.to_uppercase();
+            self.absent_columns
+                .iter()
+                .find(|c| upper.contains(&c.to_uppercase()))
+                .map(String::as_str)
+        }
     }
 
     impl OraDbEngine for MiniOra {
@@ -511,8 +530,15 @@ pub(crate) mod test_support {
             Ok(())
         }
         fn query_table(&self, query: &SqlQuery) -> QueryResult {
+            self.asked.lock().unwrap().push(query.as_str().to_owned());
+            if let Some(column) = self.missing_column_in(query.as_str()) {
+                return QueryResult(Err(anyhow::anyhow!(
+                    "ORA-00904: \"{column}\": invalid identifier"
+                )));
+            }
             let rows = match query.as_str() {
-                INSTANCE_INFO_SQL_TEXT_NEW => &self.instance_rows,
+                INSTANCE_INFO_SQL_TEXT_NEW | INSTANCE_INFO_SQL_TEXT_OLD => &self.instance_rows,
+                INSTANCE_VERSION_FULL | INSTANCE_VERSION => &self.version_rows,
                 PDB_DISCOVERY_SQL => &self.pdb_rows,
                 _ => &self.default_rows,
             };
@@ -523,16 +549,40 @@ pub(crate) mod test_support {
                 instance_rows: self.instance_rows.clone(),
                 pdb_rows: self.pdb_rows.clone(),
                 default_rows: self.default_rows.clone(),
+                absent_columns: self.absent_columns.clone(),
+                asked: Arc::clone(&self.asked),
+                version_rows: self.version_rows.clone(),
             })
         }
     }
 
     impl MiniOra {
         /// One non-CDB instance `name`; custom queries return `details:ok`.
         pub fn single(name: &str) -> Self {
+            Self::at_version(name, "19.1.0.0", "NO")
+        }
+
+        /// Derives `absent_columns` from `version`.
+        pub fn at_version(name: &str, version: &str, cdb: &str) -> Self {
+            let numeric: Vec<u32> = version
+                .split('.')
+                .filter_map(|p| p.parse::<u32>().ok())
+                .collect();
+            let major = numeric.first().copied().unwrap_or(0);
+            let minor = numeric.get(1).copied().unwrap_or(0);
+            let mut absent_columns = Vec::new();
+            if major < 18 {
+                absent_columns.push("VERSION_FULL".to_string());
+            }
+            if major < 12 || (major == 12 && minor < 1) {
+                absent_columns.push("CON_ID".to_string());
+                absent_columns.push("d.cdb".to_string());
+            }
             Self {
-                instance_rows: vec![instance_row(name, "19.1.0.0", "NO")],
+                instance_rows: vec![instance_row(name, version, cdb)],
                 default_rows: vec![vec!["details:ok".to_string()]],
+                version_rows: vec![vec![version.to_string()]],
+                absent_columns,
                 ..Default::default()
             }
         }

packages/mk-oracle/src/ora_sql/instance.rs

@@ -872,13 +872,15 @@ oracle:
         );
         let db = MiniOra {
             instance_rows: vec![instance_row("ORCL", "19.1.0.0", "NO")],
+            version_rows: vec![vec!["19.1.0.0".to_string()]],
             pdb_rows: vec![],
             default_rows: vec![
                 vec!["details:price=5; still ok".to_string()],
                 vec!["perfdata:price=5;10;20;;".to_string()],
                 vec!["long:extended detail".to_string()],
                 vec!["exit:0".to_string()],
             ],
+            ..Default::default()
         };
 
         let out = emit(
@@ -963,12 +965,14 @@ oracle:
         );
         let db = MiniOra {
             instance_rows: vec![instance_row("ORCL", "19.1.0.0", "YES")],
+            version_rows: vec![vec!["19.1.0.0".to_string()]],
             pdb_rows: vec![
                 vec!["CDB$ROOT".to_string()],
                 vec!["PDB$SEED".to_string()],
                 vec!["PDB1".to_string()],
             ],
             default_rows: vec![vec!["details:ok".to_string()]],
+            ..Default::default()
         };
 
         let out = emit(

packages/mk-oracle/src/ora_sql/sqls.rs

@@ -275,6 +275,8 @@ SELECT
     FROM v$instance i
     join v$database d
         on i.con_id = d.con_id";
+        pub const INSTANCE_VERSION_FULL: &str = r"SELECT VERSION_FULL FROM v$instance";
+        pub const INSTANCE_VERSION: &str = r"SELECT VERSION FROM v$instance";
         /// ASM instances have no mounted `v$database`.
         pub const ASM_INSTANCE_INFO_SQL_TEXT_NEW: &str = r"
 SELECT
@@ -292,8 +294,6 @@ SELECT
     INSTANCE_NAME,
     'NO'
     FROM v$instance";
-        pub const INSTANCE_APPROXIMATE_VERSION: &str =
-            r"SELECT BANNER_FULL FROM v$version WHERE banner LIKE 'Oracle%'";
         /// Lists PDBs of the connected CDB. The root container (CDB$ROOT) is
         /// not in V$PDBS, but we exclude it defensively. PDB$SEED is the
         /// read-only template and must always be filtered out.

packages/mk-oracle/src/ora_sql/system.rs

@@ -75,49 +75,87 @@ impl WorkInstances {
     }
 }
 
+/// The only error that means "this release lacks the column"; anything else is
+/// a real failure and must not be retried away.
+fn is_unknown_column_error(e: &anyhow::Error) -> bool {
+    e.to_string().contains("ORA-00904")
+}
+
+/// Oracle changed the versioning information with release 18c, introducing
+/// `VERSION_FULL`. Where that is unavailable, `VERSION` carries the full
+/// versioning information instead.
+#[derive(Debug)]
+enum DetectedVersion {
+    Since18c(InstanceVersion),
+    Before18c(InstanceVersion),
+}
+
+impl DetectedVersion {
+    fn version(&self) -> &InstanceVersion {
+        match self {
+            Self::Since18c(v) | Self::Before18c(v) => v,
+        }
+    }
+}
+
+fn _detect_version(spot: &OpenedSpot) -> Result<DetectedVersion> {
+    let ask = |sql: &str| -> Result<InstanceVersion> {
+        let rows = spot.query_table(&SqlQuery::new(sql, &Vec::new())).0?;
+        _extract_version(rows.into_iter().map(|row| row.join("")).collect())
+            .map(InstanceVersion::from)
+            .ok_or_else(|| anyhow::anyhow!("No usable version in v$instance"))
+    };
+    match ask(sqls::query::internal::INSTANCE_VERSION_FULL) {
+        Ok(version) => Ok(DetectedVersion::Since18c(version)),
+        Err(e) if is_unknown_column_error(&e) => {
+            log::info!("No VERSION_FULL in v$instance ({e}), asking for VERSION instead");
+            ask(sqls::query::internal::INSTANCE_VERSION).map(DetectedVersion::Before18c)
+        }
+        Err(e) => Err(e),
+    }
+}
+
 fn _get_instances(spot: &OpenedSpot, custom_query: Option<&str>) -> Result<_InstanceEntries> {
-    let (new_sql, old_sql) = if spot.target().is_asm() {
-        (
-            sqls::query::internal::ASM_INSTANCE_INFO_SQL_TEXT_NEW,
-            sqls::query::internal::ASM_INSTANCE_INFO_SQL_TEXT_OLD,
-        )
-    } else {
-        (
-            sqls::query::internal::INSTANCE_INFO_SQL_TEXT_NEW,
-            sqls::query::internal::INSTANCE_INFO_SQL_TEXT_OLD,
-        )
+    if let Some(query) = custom_query {
+        // Replaces the probe entirely; the caller owns the column set.
+        return Ok(_to_instance_entries(
+            spot.query_table(&SqlQuery::new(query, &Vec::new())).0?,
+        ));
+    }
+
+    let detected = _detect_version(spot)?;
+    log::info!("Instance reports {detected:?}");
+
+    let sql = match (spot.target().is_asm(), &detected) {
+        (true, DetectedVersion::Since18c(_)) => {
+            sqls::query::internal::ASM_INSTANCE_INFO_SQL_TEXT_NEW
+        }
+        (true, DetectedVersion::Before18c(_)) => {
+            sqls::query::internal::ASM_INSTANCE_INFO_SQL_TEXT_OLD
+        }
+        (false, DetectedVersion::Since18c(_)) => sqls::query::internal::INSTANCE_INFO_SQL_TEXT_NEW,
+        (false, DetectedVersion::Before18c(_)) => sqls::query::internal::INSTANCE_INFO_SQL_TEXT_OLD,
     };
-    if let Ok(result) = spot
-        .query_table(&SqlQuery::new(custom_query.unwrap_or(new_sql), &Vec::new()))
-        .0
-    {
-        Ok(_to_instance_entries(result))
-    } else {
-        let mut result = spot.query_table(&SqlQuery::new(old_sql, &Vec::new())).0?;
-        let result_with_version = spot
-            .query_table(&SqlQuery::new(
-                sqls::query::internal::INSTANCE_APPROXIMATE_VERSION,
-                &Vec::new(),
-            ))
-            .format("")?;
-        if let Some(version) = _extract_version(result_with_version) {
-            log::info!("Extracted version: {version}");
-            for r in result.iter_mut() {
-                r[2] = version.clone(); // Update the version column
-            }
+
+    let mut result = spot.query_table(&SqlQuery::new(sql, &Vec::new())).0?;
+    for r in result.iter_mut() {
+        if let Some(version_column) = r.get_mut(2) {
+            *version_column = detected.version().clone().into();
         }
-        Ok(_to_instance_entries(result))
     }
+    Ok(_to_instance_entries(result))
 }
 
 fn _extract_version(result: Vec<String>) -> Option<String> {
     if result.is_empty() {
         log::warn!("No version information found in v$instance");
         return None;
     }
-    result[0].split(' ').next_back().and_then(|s| {
-        convert_to_num_version(&InstanceVersion::from(s.to_owned())).map(|_| s.to_string())
-    })
+    result
+        .iter()
+        .flat_map(|line| line.split_whitespace())
+        .find(|token| convert_to_num_version(&InstanceVersion::from((*token).to_owned())).is_some())
+        .map(str::to_string)
 }
 
 fn _to_instance_entries(result: Vec<Vec<String>>) -> _InstanceEntries {
@@ -160,18 +198,115 @@ pub fn convert_to_num_version(version: &InstanceVersion) -> Option<InstanceNumVe
 mod tests {
     use super::*;
     use crate::config::ora_sql::Endpoint;
-    use crate::ora_sql::backend::test_support::{instance_row, MiniOra};
+    use crate::ora_sql::backend::test_support::MiniOra;
     use crate::ora_sql::backend::SpotBuilder;
 
+    /// Returns the derived (version, tenant) plus every query the run issued.
+    fn discover_at(version: &str, cdb: &str) -> (InstanceNumVersion, Tenant, Vec<String>) {
+        let db = MiniOra::at_version("ORCL", version, cdb);
+        let asked = std::sync::Arc::clone(&db.asked);
+        let spot = SpotBuilder::new()
+            .endpoint_target(&Endpoint::default())
+            .custom_engine(Box::new(db))
+            .build()
+            .unwrap()
+            .connect(None)
+            .unwrap();
+        let works = WorkInstances::new(&spot, None).expect("instance must be discovered");
+        let (v, t) = works
+            .get_info(&InstanceName::from("ORCL"))
+            .expect("ORCL must be present");
+        let queries = asked.lock().unwrap().clone();
+        (v, t, queries)
+    }
+
     #[test]
-    fn test_get_version() {
-        let simulated_spot = SpotBuilder::new()
+    fn test_18c_and_later_use_the_precise_version_and_ask_for_cdb() {
+        let (version, tenant, asked) = discover_at("19.28.0.0.0", "YES");
+        assert_eq!(version, InstanceNumVersion::from(19_28_00_00));
+        assert_eq!(tenant, Tenant::Cdb);
+        assert!(
+            asked.iter().any(|q| q.contains("VERSION_FULL")),
+            "{asked:?}"
+        );
+        assert!(asked.iter().any(|q| q.contains("d.cdb")), "{asked:?}");
+    }
+
+    #[test]
+    fn test_12c_falls_back_to_version_but_still_asks_for_cdb() {
+        // No VERSION_FULL, but CON_ID and CDB are there: the probe degrades
+        // while the instance query still reads the real tenancy.
+        let (version, tenant, asked) = discover_at("12.2.0.1.0", "YES");
+        assert_eq!(version, InstanceNumVersion::from(12_02_00_01));
+        assert_eq!(tenant, Tenant::Cdb);
+        assert!(
+            asked
+                .iter()
+                .any(|q| q.trim() == "SELECT VERSION FROM v$instance"),
+            "the probe must fall back to VERSION: {asked:?}"
+        );
+        assert!(asked.iter().any(|q| q.contains("d.cdb")), "{asked:?}");
+    }
+
+    /// The supported floor is 12.1.0.2, so a release without `V$DATABASE.CDB`
+    /// fails rather than being accommodated. Its version reaches the log first,
+    /// which is what tells an operator why.
+    #[test]
+    fn test_before_12c_fails_on_the_absent_cdb_column() {
+        let db = MiniOra::at_version("ORCL", "11.2.0.4.0", "irrelevant");
+        let spot = SpotBuilder::new()
+            .endpoint_target(&Endpoint::default())
+            .custom_engine(Box::new(db))
+            .build()
+            .unwrap()
+            .connect(None)
+            .unwrap();
+        let err = WorkInstances::new(&spot, None).expect_err("must not be accommodated");
+        assert!(err.to_string().contains("ORA-00904"), "{err}");
+    }
+
+    #[test]
+    fn test_a_real_error_is_not_treated_as_a_missing_column() {
+        // Anything but ORA-00904 must surface, or a permissions problem reads
+        // as an ancient Oracle.
+        let spot = SpotBuilder::new()
             .endpoint_target(&Endpoint::default())
             .custom_engine(Box::new(MiniOra {
-                instance_rows: vec![instance_row("free", "22.1.1.6.0", "YES")],
+                absent_columns: vec!["v$instance".to_string()],
                 ..Default::default()
             }))
             .build()
+            .unwrap()
+            .connect(None)
+            .unwrap();
+        let err = WorkInstances::new(&spot, None).expect_err("must not be swallowed");
+        assert!(err.to_string().contains("ORA-00904"), "{err}");
+    }
+
+    #[test]
+    fn test_extract_version_is_not_positional() {
+        assert_eq!(
+            _extract_version(vec!["23.26.0.24.03".to_string()]).as_deref(),
+            Some("23.26.0.24.03")
+        );
+        assert_eq!(
+            _extract_version(vec![
+                "Oracle Database 12c Enterprise Edition Release 12.1.0.2.0 - 64bit Production"
+                    .to_string()
+            ])
+            .as_deref(),
+            Some("12.1.0.2.0")
+        );
+        assert_eq!(_extract_version(vec![]), None);
+        assert_eq!(_extract_version(vec!["no version here".to_string()]), None);
+    }
+
+    #[test]
+    fn test_get_version() {
+        let simulated_spot = SpotBuilder::new()
+            .endpoint_target(&Endpoint::default())
+            .custom_engine(Box::new(MiniOra::at_version("free", "22.1.1.6.0", "YES")))
+            .build()
             .unwrap();
         let conn = simulated_spot.connect(None).unwrap();
         assert_eq!(