CA-171948: Adjust semantics of 'add_to_<map>' in Db layer

ocaml/database/db_cache_impl.ml

@@ -254,14 +254,24 @@ let process_structured_field_locked t (key,value) tblname fld objref proc_fn_sel
     let newval = match proc_fn_selector with
       | AddSet -> add_to_set key existing_str
       | RemoveSet -> remove_from_set key existing_str
-      | AddMap ->
+      | AddMap | AddMapLegacy ->
         begin
           try
-            add_to_map key value existing_str
+            (* We use the idempotent map add if we're using the non-legacy
+               process function, or if the global field 'idempotent_map' has
+               been set. By default, the Db calls on the master use the
+               legacy functions, but those on the slave use the new one.
+               This means xapi code should always assume idempotent_map is
+               true *)
+            let idempotent =
+              (proc_fn_selector = AddMap) || !Db_globs.idempotent_map
+            in
+            add_to_map ~idempotent key value existing_str
           with Duplicate ->
             error "Duplicate key in set or map: table %s; field %s; ref %s; key %s" tblname fld objref key;
             raise (Duplicate_key (tblname,fld,objref,key));
         end
+
       | RemoveMap -> remove_from_map key existing_str in
     write_field_locked t tblname objref fld newval
   with Not_found ->

ocaml/database/db_cache_types.ml

@@ -350,10 +350,10 @@ let remove_from_set key t =
   Schema.Value.Set (List.filter (fun x -> x <> key) t)
 
 exception Duplicate
-let add_to_map key value t =
+let add_to_map ~idempotent key value t =
   let t = Schema.Value.Unsafe_cast.pairs t in
-  if List.mem key (List.map fst t) then raise Duplicate;
-  Schema.Value.Pairs ((key, value) :: t)
+  if List.mem_assoc key t && (not idempotent || List.assoc key t <> value) then raise Duplicate;
+  Schema.Value.Pairs ((key, value) :: List.filter (fun (k, _) -> k <> key) t)
 
 let remove_from_map key t =
   let t = Schema.Value.Unsafe_cast.pairs t in
@@ -500,4 +500,5 @@ type structured_op_t =
   | RemoveSet
   | AddMap
   | RemoveMap
+  | AddMapLegacy
 [@@deriving rpc]

ocaml/database/db_cache_types.mli

@@ -146,7 +146,7 @@ end
 exception Duplicate
 val add_to_set : string -> Schema.Value.t -> Schema.Value.t
 val remove_from_set : string -> Schema.Value.t -> Schema.Value.t
-val add_to_map : string -> string -> Schema.Value.t -> Schema.Value.t
+val add_to_map : idempotent:bool -> string -> string -> Schema.Value.t -> Schema.Value.t
 val remove_from_map : string -> Schema.Value.t -> Schema.Value.t
 
 val set_field : string -> string -> string -> Schema.Value.t -> Database.t -> Database.t
@@ -169,5 +169,6 @@ type structured_op_t =
   | RemoveSet
   | AddMap
   | RemoveMap
+  | AddMapLegacy
 val structured_op_t_of_rpc: Rpc.t -> structured_op_t
 val rpc_of_structured_op_t: structured_op_t -> Rpc.t

ocaml/database/db_globs.ml

@@ -50,6 +50,9 @@ let static_vdis_dir = ref "/etc/xensource/static-vdis"
 (* Note the following has an equivalent in the xapi layer *)
 let http_limit_max_rpc_size = 300 * 1024 (* 300K *)
 
+(* add_to_map is idempotent *)
+let idempotent_map = ref false
+
 (** Dynamic configurations to be read whenever xapi (re)start *)
 
 let permanent_master_failure_retry_interval = ref 60.

ocaml/database/db_rpc_common_v1.ml

@@ -79,7 +79,9 @@ let marshall_structured_op x =
       AddSet -> "addset"
     | RemoveSet -> "removeset"
     | AddMap -> "addmap"
-    | RemoveMap -> "removemap" in
+    | RemoveMap -> "removemap"
+    | AddMapLegacy -> "addmap" (* Nb, we always use 'non-legacy' mode for remote access *)
+  in
   XMLRPC.To.string str
 let unmarshall_structured_op xml =
   match (XMLRPC.From.string xml) with
@@ -311,4 +313,3 @@ let unmarshall_read_records_where_response xml =
            [ref_xml; rec_xml] -> (XMLRPC.From.string ref_xml, unmarshall_read_record_response rec_xml)
          | _ -> raise DB_remote_marshall_error)
       xml_refs_and_recs_list
-

ocaml/database/db_rpc_common_v2.ml

@@ -34,6 +34,16 @@ module Request = struct
     | Read_records_where of string * Db_filter_types.expr
     | Process_structured_field of (string * string) * string * string * string * Db_cache_types.structured_op_t
   [@@deriving rpc]
+
+  (* Make sure the slave only ever uses the idempotent version *)
+  let rpc_of_t t =
+    let t' =
+      match t with
+      | Process_structured_field (a,b,c,d,Db_cache_types.AddMapLegacy) ->
+        Process_structured_field (a,b,c,d,Db_cache_types.AddMap)
+      | x -> x
+    in
+    rpc_of_t t'
 end
 
 module Response = struct

ocaml/idl/ocaml_backend/gen_db_actions.ml

@@ -312,7 +312,7 @@ let db_action api : O.Module.t =
           (Escaping.escape_id full_name)
           Client._self
       | FromField(Add, { DT.ty = DT.Map(_, _); full_name = full_name }) ->
-        Printf.sprintf "DB.process_structured_field __t (%s,%s) \"%s\" \"%s\" %s AddMap"
+        Printf.sprintf "DB.process_structured_field __t (%s,%s) \"%s\" \"%s\" %s AddMapLegacy"
           Client._key Client._value
           (Escaping.escape_obj obj.DT.name)
           (Escaping.escape_id full_name)
@@ -472,4 +472,3 @@ let db_defaults api : O.Signature.t =
   { O.Signature.name = _db_defaults;
     elements = List.map (fun x -> O.Signature.Module (obj x)) (Dm_api.objects_of_api api)
   }
-

ocaml/xapi/test_db_lowlevel.ml

@@ -41,8 +41,38 @@ let test_db_get_all_records_race () =
 let tear_down () =
   Db_cache_impl.fist_delay_read_records_where := false
 
+let test_idempotent_map () =
+  Db_globs.idempotent_map := false;
+  let __context = make_test_database () in
+  let (vm_ref: API.ref_VM) = make_vm ~__context () in
+  Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value";
+  assert_raises (Db_exn.Duplicate_key ("VM","other_config",(Ref.string_of vm_ref),"test"))
+    (fun () -> Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value");
+  assert_raises (Db_exn.Duplicate_key ("VM","other_config",(Ref.string_of vm_ref),"test"))
+    (fun () -> Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value2");
+
+  Db_globs.idempotent_map := true;
+  let __context = make_test_database () in
+  let (vm_ref: API.ref_VM) = make_vm ~__context () in
+  Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value";
+  assert_equal (Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value") ();
+  assert_raises (Db_exn.Duplicate_key ("VM","other_config",(Ref.string_of vm_ref),"test"))
+    (fun () -> Db.VM.add_to_other_config ~__context ~self:vm_ref ~key:"test" ~value:"value2");
+
+  Db_globs.idempotent_map := false
+
+let test_slave_uses_nonlegacy_addmap () =
+  let operation = Db_cache_types.AddMapLegacy in
+  let operation' = Db_rpc_common_v1.marshall_structured_op operation |> Db_rpc_common_v1.unmarshall_structured_op in
+  assert_equal operation' Db_cache_types.AddMap;
+  let operationv2 = Db_rpc_common_v2.Request.Process_structured_field (("",""),"","","",Db_cache_types.AddMapLegacy) in
+  let operationv2' = Db_rpc_common_v2.Request.(operationv2 |> rpc_of_t |> t_of_rpc) in
+  assert_equal operationv2' (Db_rpc_common_v2.Request.Process_structured_field (("",""),"","","",Db_cache_types.AddMap))
+
 let test =
   "test_db_lowlevel" >:::
   [
     "test_db_get_all_records_race" >:: (bracket id test_db_get_all_records_race tear_down);
+    "test_db_idempotent_map" >:: test_idempotent_map;
+    "test_slaves_use_nonlegacy_addmap" >:: test_slave_uses_nonlegacy_addmap;
   ]

ocaml/xapi/xapi_globs.ml

@@ -980,6 +980,10 @@ let other_options = [
 
   "modprobe_path", Arg.Set_string modprobe_path,
   (fun () -> !modprobe_path), "Location of the modprobe(8) command: should match $(which modprobe)";
+
+  "db_idempotent_map", Arg.Set Db_globs.idempotent_map,
+  (fun () -> string_of_bool !Db_globs.idempotent_map), "True if the add_to_<map> API calls should be idempotent";
+
 ]
 
 let all_options = options_of_xapi_globs_spec @ other_options