1- #![ allow( dead_code) ]
21use anyhow:: Result ;
3- use common:: Node ;
4- use etcd_client:: { Client , GetOptions , PutOptions } ;
5- use serde_yaml;
2+ use etcd_client:: { Client , GetOptions , PutOptions , WatchOptions , WatchStream , Watcher } ;
63use std:: sync:: Arc ;
74use tokio:: sync:: RwLock ;
85
96use crate :: protocol:: config:: NetworkConfig ;
10- /// like etcd, k:/registry/pods/pod_name v:yaml file of pod
11- /// k:/registry/nodes/node_name v:yaml file of node
7+
8+ /// XlineStore provides an etcd-like API for managing pods and nodes.
9+ /// Keys are stored under `/registry/pods/` and `/registry/nodes/`.
10+ /// Values are YAML serialized definitions.
1211#[ derive( Clone ) ]
1312pub struct XlineStore {
1413 client : Arc < RwLock < Client > > ,
1514}
1615
16+ #[ allow( unused) ]
1717impl XlineStore {
18+ /// Create a new XlineStore instance by connecting to the given endpoints.
1819pub async fn new ( endpoints : & [ & str ] ) -> Result < Self > {
1920 let client = Client :: connect ( endpoints, None ) . await ?;
2021 Ok ( Self {
2122 client : Arc :: new ( RwLock :: new ( client) ) ,
2223 } )
2324 }
2425
25- pub async fn insert_pod_yaml ( & self , pod_name : & str , pod_yaml : & str ) -> Result < ( ) > {
26- let key = format ! ( "/registry/pods/{pod_name}" ) ;
27- let mut client = self . client . write ( ) . await ;
28- client. put ( key, pod_yaml, None ) . await ?;
29- Ok ( ( ) )
26+ /// Get a read-only reference to the internal etcd client.
27+ /// This is typically used for watch operations.
28+ pub async fn client ( & self ) -> tokio:: sync:: RwLockReadGuard < ' _ , Client > {
29+ self . client . read ( ) . await
3030 }
3131
32- pub async fn get_pod_yaml ( & self , pod_name : & str ) -> Result < Option < String > > {
33- let key = format ! ( "/registry/pods/{pod_name}" ) ;
32+ /// List all pod names (keys only, values are ignored).
33+ pub async fn list_pods ( & self ) -> Result < Vec < String > > {
34+ let key = "/registry/pods/" . to_string ( ) ;
3435 let mut client = self . client . write ( ) . await ;
35- let resp = client. get ( key, None ) . await ?;
36+ let resp = client
37+ . get (
38+ key. clone ( ) ,
39+ Some ( GetOptions :: new ( ) . with_prefix ( ) . with_keys_only ( ) ) ,
40+ )
41+ . await ?;
3642 Ok ( resp
3743 . kvs ( )
38- . first ( )
39- . map ( |kv| String :: from_utf8_lossy ( kv. value ( ) ) . to_string ( ) ) )
44+ . iter ( )
45+ . map ( |kv| String :: from_utf8_lossy ( kv. key ( ) ) . replace ( "/registry/pods/" , "" ) )
46+ . collect ( ) )
4047 }
4148
42- pub async fn list_pods ( & self ) -> Result < Vec < String > > {
49+ /// List all node names (keys only, values are ignored).
50+ pub async fn list_nodes ( & self ) -> Result < Vec < String > > {
51+ let key = "/registry/nodes/" . to_string ( ) ;
4352 let mut client = self . client . write ( ) . await ;
4453 let resp = client
45- . get ( "/registry/pods/" , Some ( GetOptions :: new ( ) . with_prefix ( ) ) )
54+ . get (
55+ key. clone ( ) ,
56+ Some ( GetOptions :: new ( ) . with_prefix ( ) . with_keys_only ( ) ) ,
57+ )
4658 . await ?;
4759 Ok ( resp
4860 . kvs ( )
4961 . iter ( )
50- . map ( |kv| String :: from_utf8_lossy ( kv. key ( ) ) . replace ( "/registry/pods /" , "" ) )
62+ . map ( |kv| String :: from_utf8_lossy ( kv. key ( ) ) . replace ( "/registry/nodes /" , "" ) )
5163 . collect ( ) )
5264 }
5365
66+ /// Insert a node YAML definition into xline.
5467pub async fn insert_node_yaml ( & self , node_name : & str , node_yaml : & str ) -> Result < ( ) > {
5568 let key = format ! ( "/registry/nodes/{node_name}" ) ;
5669 let mut client = self . client . write ( ) . await ;
5770 client. put ( key, node_yaml, Some ( PutOptions :: new ( ) ) ) . await ?;
5871 Ok ( ( ) )
5972 }
6073
74+ // Example (currently unused):
6175 // pub async fn get_node_yaml(&self, node_name: &str) -> Result<Option<String>> {
6276 // let key = format!("/registry/nodes/{node_name}");
6377 // let mut client = self.client.write().await;
@@ -74,20 +88,27 @@ impl XlineStore {
7488 // }
7589 // }
7690
77- pub async fn list_nodes ( & self ) -> Result < Vec < ( String , Node ) > > {
91+ /// Insert a pod YAML definition into xline.
92+ pub async fn insert_pod_yaml ( & self , pod_name : & str , pod_yaml : & str ) -> Result < ( ) > {
93+ let key = format ! ( "/registry/pods/{pod_name}" ) ;
7894 let mut client = self . client . write ( ) . await ;
79- let resp = client
80- . get ( "/registry/nodes/" , Some ( GetOptions :: new ( ) . with_prefix ( ) ) )
81- . await ?;
82- let mut result = Vec :: new ( ) ;
83- for kv in resp. kvs ( ) {
84- let name = String :: from_utf8_lossy ( kv. key ( ) ) . replace ( "/registry/nodes/" , "" ) ;
85- let node: Node = serde_yaml:: from_slice ( kv. value ( ) ) ?;
86- result. push ( ( name, node) ) ;
95+ client. put ( key, pod_yaml, Some ( PutOptions :: new ( ) ) ) . await ?;
96+ Ok ( ( ) )
97+ }
98+
99+ /// Get a pod YAML definition from xline.
100+ pub async fn get_pod_yaml ( & self , pod_name : & str ) -> Result < Option < String > > {
101+ let key = format ! ( "/registry/pods/{pod_name}" ) ;
102+ let mut client = self . client . write ( ) . await ;
103+ let resp = client. get ( key, None ) . await ?;
104+ if let Some ( kv) = resp. kvs ( ) . first ( ) {
105+ Ok ( Some ( String :: from_utf8_lossy ( kv. value ( ) ) . to_string ( ) ) )
106+ } else {
107+ Ok ( None )
87108 }
88- Ok ( result)
89109 }
90110
111+ /// Delete a pod from xline.
91112pub async fn delete_pod ( & self , pod_name : & str ) -> Result < ( ) > {
92113 let key = format ! ( "/registry/pods/{pod_name}" ) ;
93114 let mut client = self . client . write ( ) . await ;
@@ -123,4 +144,56 @@ impl XlineStore {
123144 Ok ( None )
124145 }
125146 }
147+
148+ /// Take a snapshot of all pods and return them with the current revision.
149+ pub async fn pods_snapshot_with_rev ( & self ) -> Result < ( Vec < ( String , String ) > , i64 ) > {
150+ let key_prefix = "/registry/pods/" . to_string ( ) ;
151+ let mut client = self . client . write ( ) . await ;
152+ let resp = client
153+ . get ( key_prefix. clone ( ) , Some ( GetOptions :: new ( ) . with_prefix ( ) ) )
154+ . await ?;
155+ let rev = resp. header ( ) . map ( |h| h. revision ( ) ) . unwrap_or ( 0 ) ;
156+ let items: Vec < ( String , String ) > = resp
157+ . kvs ( )
158+ . iter ( )
159+ . map ( |kv| {
160+ (
161+ String :: from_utf8_lossy ( kv. key ( ) ) . replace ( "/registry/pods/" , "" ) ,
162+ String :: from_utf8_lossy ( kv. value ( ) ) . to_string ( ) ,
163+ )
164+ } )
165+ . collect ( ) ;
166+ Ok ( ( items, rev) )
167+ }
168+
169+ /// Create a watch on all pods with prefix `/registry/pods/`, starting from a given revision.
170+ pub async fn watch_pods ( & self , start_rev : i64 ) -> Result < ( Watcher , WatchStream ) > {
171+ let key_prefix = "/registry/pods/" . to_string ( ) ;
172+ let opts = WatchOptions :: new ( )
173+ . with_prefix ( )
174+ . with_prev_key ( )
175+ . with_start_revision ( start_rev) ;
176+ let mut client = self . client . write ( ) . await ;
177+ let ( watcher, stream) = client. watch ( key_prefix, Some ( opts) ) . await ?;
178+ Ok ( ( watcher, stream) )
179+ }
180+
181+ /// Initialize Flannel CNI network configuration.
182+ pub async fn init_flannel_config ( & self ) -> Result < ( ) > {
183+ let config_json = r#"{
184+ "Network": "10.244.0.0/16",
185+ "SubnetLen": 24,
186+ "Backend": {
187+ "Type": "vxlan",
188+ "VNI": 1
189+ }
190+ }"# ;
191+
192+ let key = "/coreos.com/network/config" ;
193+ let mut client = self . client . write ( ) . await ;
194+ client
195+ . put ( key, config_json, Some ( PutOptions :: new ( ) ) )
196+ . await ?;
197+ Ok ( ( ) )
198+ }
126199}
0 commit comments