| OLD | NEW |
| 1 // Copyright 2016 The LUCI Authors. All rights reserved. | 1 // Copyright 2016 The LUCI Authors. All rights reserved. |
| 2 // Use of this source code is governed under the Apache License, Version 2.0 | 2 // Use of this source code is governed under the Apache License, Version 2.0 |
| 3 // that can be found in the LICENSE file. | 3 // that can be found in the LICENSE file. |
| 4 | 4 |
| 5 package cloud | 5 package cloud |
| 6 | 6 |
| 7 import ( | 7 import ( |
| 8 "fmt" | 8 "fmt" |
| 9 "reflect" | 9 "reflect" |
| 10 "strings" | 10 "strings" |
| 11 "time" | 11 "time" |
| 12 | 12 |
| 13 "github.com/luci/luci-go/common/errors" | 13 "github.com/luci/luci-go/common/errors" |
| 14 | 14 |
| 15 ds "github.com/luci/gae/service/datastore" |
| 16 "github.com/luci/gae/service/info" |
| 17 |
| 15 "cloud.google.com/go/datastore" | 18 "cloud.google.com/go/datastore" |
| 16 ds "github.com/luci/gae/service/datastore" | |
| 17 infoS "github.com/luci/gae/service/info" | |
| 18 | 19 |
| 19 "golang.org/x/net/context" | 20 "golang.org/x/net/context" |
| 20 ) | 21 ) |
| 21 | 22 |
| 22 type cloudDatastore struct { | 23 type cloudDatastore struct { |
| 23 client *datastore.Client | 24 client *datastore.Client |
| 24 } | 25 } |
| 25 | 26 |
| 26 func (cds *cloudDatastore) use(c context.Context) context.Context { | 27 func (cds *cloudDatastore) use(c context.Context) context.Context { |
| 27 » return ds.SetRawFactory(c, func(ic context.Context, wantTxn bool) ds.Raw
Interface { | 28 » return ds.SetRawFactory(c, func(ic context.Context) ds.RawInterface { |
| 28 » » inf := infoS.Get(ic) | 29 » » if ns := info.GetNamespace(ic); ns != "" { |
| 29 » » if ns, ok := inf.GetNamespace(); ok { | |
| 30 ic = datastore.WithNamespace(ic, ns) | 30 ic = datastore.WithNamespace(ic, ns) |
| 31 } | 31 } |
| 32 | 32 |
| 33 » » bds := boundDatastore{ | 33 » » return &boundDatastore{ |
| 34 Context: ic, | 34 Context: ic, |
| 35 cloudDatastore: cds, | 35 cloudDatastore: cds, |
| 36 » » » appID: inf.FullyQualifiedAppID(), | 36 » » » transaction: datastoreTransaction(ic), |
| 37 » » » kc: ds.GetKeyContext(ic), |
| 37 } | 38 } |
| 38 if wantTxn { | |
| 39 bds.transaction = datastoreTransaction(ic) | |
| 40 } | |
| 41 return &bds | |
| 42 }) | 39 }) |
| 43 } | 40 } |
| 44 | 41 |
| 45 // boundDatastore is a bound instance of the cloudDatastore installed in the | 42 // boundDatastore is a bound instance of the cloudDatastore installed in the |
| 46 // Context. | 43 // Context. |
| 47 type boundDatastore struct { | 44 type boundDatastore struct { |
| 45 context.Context |
| 46 |
| 48 // Context is the bound user Context. It includes the datastore namespac
e, if | 47 // Context is the bound user Context. It includes the datastore namespac
e, if |
| 49 // one is set. | 48 // one is set. |
| 50 context.Context | |
| 51 *cloudDatastore | 49 *cloudDatastore |
| 52 | 50 |
| 53 appID string | |
| 54 transaction *datastore.Transaction | 51 transaction *datastore.Transaction |
| 52 kc ds.KeyContext |
| 55 } | 53 } |
| 56 | 54 |
| 57 func (bds *boundDatastore) AllocateIDs(keys []*ds.Key, cb ds.NewKeyCB) error { | 55 func (bds *boundDatastore) AllocateIDs(keys []*ds.Key, cb ds.NewKeyCB) error { |
| 58 nativeKeys, err := bds.client.AllocateIDs(bds, bds.gaeKeysToNative(keys.
..)) | 56 nativeKeys, err := bds.client.AllocateIDs(bds, bds.gaeKeysToNative(keys.
..)) |
| 59 if err != nil { | 57 if err != nil { |
| 60 return normalizeError(err) | 58 return normalizeError(err) |
| 61 } | 59 } |
| 62 | 60 |
| 63 keys = bds.nativeKeysToGAE(nativeKeys...) | 61 keys = bds.nativeKeysToGAE(nativeKeys...) |
| 64 for _, key := range keys { | 62 for _, key := range keys { |
| (...skipping 47 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 112 npls = bds.mkNPLS(nil) | 110 npls = bds.mkNPLS(nil) |
| 113 } | 111 } |
| 114 nativeKey, err := it.Next(npls) | 112 nativeKey, err := it.Next(npls) |
| 115 if err != nil { | 113 if err != nil { |
| 116 if err == datastore.Done { | 114 if err == datastore.Done { |
| 117 return nil | 115 return nil |
| 118 } | 116 } |
| 119 return normalizeError(err) | 117 return normalizeError(err) |
| 120 } | 118 } |
| 121 | 119 |
| 122 » » if err := cb(bds.nativeKeysToGAE(nativeKey)[0], npls.pmap, curso
rFn); err != nil { | 120 » » var pmap ds.PropertyMap |
| 121 » » if npls != nil { |
| 122 » » » pmap = npls.pmap |
| 123 » » } |
| 124 » » if err := cb(bds.nativeKeysToGAE(nativeKey)[0], pmap, cursorFn);
err != nil { |
| 123 if err == ds.Stop { | 125 if err == ds.Stop { |
| 124 return nil | 126 return nil |
| 125 } | 127 } |
| 126 return normalizeError(err) | 128 return normalizeError(err) |
| 127 } | 129 } |
| 128 } | 130 } |
| 129 } | 131 } |
| 130 | 132 |
| 131 func (bds *boundDatastore) Count(q *ds.FinalizedQuery) (int64, error) { | 133 func (bds *boundDatastore) Count(q *ds.FinalizedQuery) (int64, error) { |
| 132 v, err := bds.client.Count(bds, bds.prepareNativeQuery(q)) | 134 v, err := bds.client.Count(bds, bds.prepareNativeQuery(q)) |
| (...skipping 26 matching lines...) Expand all Loading... |
| 159 } | 161 } |
| 160 | 162 |
| 161 func (bds *boundDatastore) GetMulti(keys []*ds.Key, _meta ds.MultiMetaGetter, cb
ds.GetMultiCB) error { | 163 func (bds *boundDatastore) GetMulti(keys []*ds.Key, _meta ds.MultiMetaGetter, cb
ds.GetMultiCB) error { |
| 162 nativeKeys := bds.gaeKeysToNative(keys...) | 164 nativeKeys := bds.gaeKeysToNative(keys...) |
| 163 nativePLS := make([]*nativePropertyLoadSaver, len(nativeKeys)) | 165 nativePLS := make([]*nativePropertyLoadSaver, len(nativeKeys)) |
| 164 for i := range nativePLS { | 166 for i := range nativePLS { |
| 165 nativePLS[i] = bds.mkNPLS(nil) | 167 nativePLS[i] = bds.mkNPLS(nil) |
| 166 } | 168 } |
| 167 | 169 |
| 168 var err error | 170 var err error |
| 169 » if tx := bds.transaction; tx != nil { | 171 » if bds.transaction != nil { |
| 170 // Transactional GetMulti. | 172 // Transactional GetMulti. |
| 171 » » err = tx.GetMulti(nativeKeys, nativePLS) | 173 » » err = bds.transaction.GetMulti(nativeKeys, nativePLS) |
| 172 } else { | 174 } else { |
| 173 // Non-transactional GetMulti. | 175 // Non-transactional GetMulti. |
| 174 err = bds.client.GetMulti(bds, nativeKeys, nativePLS) | 176 err = bds.client.GetMulti(bds, nativeKeys, nativePLS) |
| 175 } | 177 } |
| 176 | 178 |
| 177 return idxCallbacker(err, len(nativePLS), func(idx int, err error) error
{ | 179 return idxCallbacker(err, len(nativePLS), func(idx int, err error) error
{ |
| 178 return cb(nativePLS[idx].pmap, err) | 180 return cb(nativePLS[idx].pmap, err) |
| 179 }) | 181 }) |
| 180 } | 182 } |
| 181 | 183 |
| 182 func (bds *boundDatastore) PutMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds
.NewKeyCB) error { | 184 func (bds *boundDatastore) PutMulti(keys []*ds.Key, vals []ds.PropertyMap, cb ds
.NewKeyCB) error { |
| 183 nativeKeys := bds.gaeKeysToNative(keys...) | 185 nativeKeys := bds.gaeKeysToNative(keys...) |
| 184 nativePLS := make([]*nativePropertyLoadSaver, len(vals)) | 186 nativePLS := make([]*nativePropertyLoadSaver, len(vals)) |
| 185 for i := range nativePLS { | 187 for i := range nativePLS { |
| 186 nativePLS[i] = bds.mkNPLS(vals[i]) | 188 nativePLS[i] = bds.mkNPLS(vals[i]) |
| 187 } | 189 } |
| 188 | 190 |
| 189 var err error | 191 var err error |
| 190 » if tx := bds.transaction; tx != nil { | 192 » if bds.transaction != nil { |
| 191 // Transactional PutMulti. | 193 // Transactional PutMulti. |
| 192 // | 194 // |
| 193 // In order to simulate the presence of mid-transaction key allo
cation, we | 195 // In order to simulate the presence of mid-transaction key allo
cation, we |
| 194 // will identify any incomplete keys and allocate IDs for them.
This is | 196 // will identify any incomplete keys and allocate IDs for them.
This is |
| 195 // potentially wasteful in the event of failed or retried transa
ctions, but | 197 // potentially wasteful in the event of failed or retried transa
ctions, but |
| 196 // it is required to maintain API compatibility with the datasto
re | 198 // it is required to maintain API compatibility with the datasto
re |
| 197 // interface. | 199 // interface. |
| 198 var incompleteKeys []*datastore.Key | 200 var incompleteKeys []*datastore.Key |
| 199 var incompleteKeyMap map[int]int | 201 var incompleteKeyMap map[int]int |
| 200 for i, k := range nativeKeys { | 202 for i, k := range nativeKeys { |
| (...skipping 11 matching lines...) Expand all Loading... |
| 212 if len(incompleteKeys) > 0 { | 214 if len(incompleteKeys) > 0 { |
| 213 idKeys, err := bds.client.AllocateIDs(bds, incompleteKey
s) | 215 idKeys, err := bds.client.AllocateIDs(bds, incompleteKey
s) |
| 214 if err != nil { | 216 if err != nil { |
| 215 return err | 217 return err |
| 216 } | 218 } |
| 217 for i, idKey := range idKeys { | 219 for i, idKey := range idKeys { |
| 218 nativeKeys[incompleteKeyMap[i]] = idKey | 220 nativeKeys[incompleteKeyMap[i]] = idKey |
| 219 } | 221 } |
| 220 } | 222 } |
| 221 | 223 |
| 222 » » _, err = tx.PutMulti(nativeKeys, nativePLS) | 224 » » _, err = bds.transaction.PutMulti(nativeKeys, nativePLS) |
| 223 } else { | 225 } else { |
| 224 // Non-transactional PutMulti. | 226 // Non-transactional PutMulti. |
| 225 nativeKeys, err = bds.client.PutMulti(bds, nativeKeys, nativePLS
) | 227 nativeKeys, err = bds.client.PutMulti(bds, nativeKeys, nativePLS
) |
| 226 } | 228 } |
| 227 | 229 |
| 228 return idxCallbacker(err, len(nativeKeys), func(idx int, err error) erro
r { | 230 return idxCallbacker(err, len(nativeKeys), func(idx int, err error) erro
r { |
| 229 if err == nil { | 231 if err == nil { |
| 230 return cb(bds.nativeKeysToGAE(nativeKeys[idx])[0], nil) | 232 return cb(bds.nativeKeysToGAE(nativeKeys[idx])[0], nil) |
| 231 } | 233 } |
| 232 return cb(nil, err) | 234 return cb(nil, err) |
| 233 }) | 235 }) |
| 234 } | 236 } |
| 235 | 237 |
| 236 func (bds *boundDatastore) DeleteMulti(keys []*ds.Key, cb ds.DeleteMultiCB) erro
r { | 238 func (bds *boundDatastore) DeleteMulti(keys []*ds.Key, cb ds.DeleteMultiCB) erro
r { |
| 237 nativeKeys := bds.gaeKeysToNative(keys...) | 239 nativeKeys := bds.gaeKeysToNative(keys...) |
| 238 | 240 |
| 239 var err error | 241 var err error |
| 240 » if tx := bds.transaction; tx != nil { | 242 » if bds.transaction != nil { |
| 241 // Transactional DeleteMulti. | 243 // Transactional DeleteMulti. |
| 242 » » err = tx.DeleteMulti(nativeKeys) | 244 » » err = bds.transaction.DeleteMulti(nativeKeys) |
| 243 } else { | 245 } else { |
| 244 // Non-transactional DeleteMulti. | 246 // Non-transactional DeleteMulti. |
| 245 err = bds.client.DeleteMulti(bds, nativeKeys) | 247 err = bds.client.DeleteMulti(bds, nativeKeys) |
| 246 } | 248 } |
| 247 | 249 |
| 248 return idxCallbacker(err, len(nativeKeys), func(_ int, err error) error
{ | 250 return idxCallbacker(err, len(nativeKeys), func(_ int, err error) error
{ |
| 249 return cb(err) | 251 return cb(err) |
| 250 }) | 252 }) |
| 251 } | 253 } |
| 252 | 254 |
| 253 func (bds *boundDatastore) Testable() ds.Testable { | 255 func (bds *boundDatastore) WithoutTransaction() context.Context { |
| 254 » return nil | 256 » return withDatastoreTransaction(bds, nil) |
| 255 } | 257 } |
| 256 | 258 |
| 259 func (bds *boundDatastore) CurrentTransaction() ds.Transaction { return bds.tran
saction } |
| 260 |
| 261 func (bds *boundDatastore) GetTestable() ds.Testable { return nil } |
| 262 |
| 257 func (bds *boundDatastore) prepareNativeQuery(fq *ds.FinalizedQuery) *datastore.
Query { | 263 func (bds *boundDatastore) prepareNativeQuery(fq *ds.FinalizedQuery) *datastore.
Query { |
| 258 nq := datastore.NewQuery(fq.Kind()) | 264 nq := datastore.NewQuery(fq.Kind()) |
| 259 if bds.transaction != nil { | 265 if bds.transaction != nil { |
| 260 nq = nq.Transaction(bds.transaction) | 266 nq = nq.Transaction(bds.transaction) |
| 261 } | 267 } |
| 262 | 268 |
| 263 // nativeFilter translates a filter field. If the translation fails, we'
ll | 269 // nativeFilter translates a filter field. If the translation fails, we'
ll |
| 264 // pass the result through to the underlying datastore and allow it to | 270 // pass the result through to the underlying datastore and allow it to |
| 265 // reject it. | 271 // reject it. |
| 266 nativeFilter := func(prop ds.Property) interface{} { | 272 nativeFilter := func(prop ds.Property) interface{} { |
| (...skipping 56 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 323 if ic.Descending { | 329 if ic.Descending { |
| 324 prop = "-" + prop | 330 prop = "-" + prop |
| 325 } | 331 } |
| 326 nq = nq.Order(prop) | 332 nq = nq.Order(prop) |
| 327 } | 333 } |
| 328 | 334 |
| 329 return nq | 335 return nq |
| 330 } | 336 } |
| 331 | 337 |
| 332 func (bds *boundDatastore) mkNPLS(base ds.PropertyMap) *nativePropertyLoadSaver
{ | 338 func (bds *boundDatastore) mkNPLS(base ds.PropertyMap) *nativePropertyLoadSaver
{ |
| 333 » return &nativePropertyLoadSaver{bds: bds, pmap: clonePropertyMap(base)} | 339 » return &nativePropertyLoadSaver{ |
| 340 » » bds: bds, |
| 341 » » pmap: clonePropertyMap(base), |
| 342 » } |
| 334 } | 343 } |
| 335 | 344 |
| 336 func (bds *boundDatastore) gaePropertyToNative(name string, pdata ds.PropertyDat
a) (nativeProp datastore.Property, err error) { | 345 func (bds *boundDatastore) gaePropertyToNative(name string, pdata ds.PropertyDat
a) ( |
| 346 » nativeProp datastore.Property, err error) { |
| 347 |
| 337 nativeProp.Name = name | 348 nativeProp.Name = name |
| 338 | 349 |
| 339 convert := func(prop *ds.Property) (interface{}, error) { | 350 convert := func(prop *ds.Property) (interface{}, error) { |
| 340 switch pt := prop.Type(); pt { | 351 switch pt := prop.Type(); pt { |
| 341 case ds.PTNull, ds.PTInt, ds.PTTime, ds.PTBool, ds.PTBytes, ds.P
TString, ds.PTFloat: | 352 case ds.PTNull, ds.PTInt, ds.PTTime, ds.PTBool, ds.PTBytes, ds.P
TString, ds.PTFloat: |
| 342 return prop.Value(), nil | 353 return prop.Value(), nil |
| 343 | 354 |
| 344 case ds.PTKey: | 355 case ds.PTKey: |
| 345 return bds.gaeKeysToNative(prop.Value().(*ds.Key))[0], n
il | 356 return bds.gaeKeysToNative(prop.Value().(*ds.Key))[0], n
il |
| 346 | 357 |
| (...skipping 28 matching lines...) Expand all Loading... |
| 375 } | 386 } |
| 376 nativeProp.Value = multiProp | 387 nativeProp.Value = multiProp |
| 377 | 388 |
| 378 default: | 389 default: |
| 379 err = fmt.Errorf("unsupported PropertyData type for %q: %T", nam
e, pdata) | 390 err = fmt.Errorf("unsupported PropertyData type for %q: %T", nam
e, pdata) |
| 380 } | 391 } |
| 381 | 392 |
| 382 return | 393 return |
| 383 } | 394 } |
| 384 | 395 |
| 385 func (bds *boundDatastore) nativePropertyToGAE(nativeProp datastore.Property) (n
ame string, pdata ds.PropertyData, err error) { | 396 func (bds *boundDatastore) nativePropertyToGAE(nativeProp datastore.Property) ( |
| 397 » name string, pdata ds.PropertyData, err error) { |
| 398 |
| 386 name = nativeProp.Name | 399 name = nativeProp.Name |
| 387 | 400 |
| 388 convert := func(nv interface{}, prop *ds.Property) error { | 401 convert := func(nv interface{}, prop *ds.Property) error { |
| 389 switch nvt := nv.(type) { | 402 switch nvt := nv.(type) { |
| 390 case nil: | 403 case nil: |
| 391 nv = nil | 404 nv = nil |
| 392 | 405 |
| 393 case int64, bool, string, float64, []byte: | 406 case int64, bool, string, float64, []byte: |
| 394 break | 407 break |
| 395 | 408 |
| (...skipping 52 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 448 nativeKey = datastore.NewKey(bds, tok.Kind, tok.StringID
, tok.IntID, nativeKey) | 461 nativeKey = datastore.NewKey(bds, tok.Kind, tok.StringID
, tok.IntID, nativeKey) |
| 449 } | 462 } |
| 450 nativeKeys[i] = nativeKey | 463 nativeKeys[i] = nativeKey |
| 451 } | 464 } |
| 452 return nativeKeys | 465 return nativeKeys |
| 453 } | 466 } |
| 454 | 467 |
| 455 func (bds *boundDatastore) nativeKeysToGAE(nativeKeys ...*datastore.Key) []*ds.K
ey { | 468 func (bds *boundDatastore) nativeKeysToGAE(nativeKeys ...*datastore.Key) []*ds.K
ey { |
| 456 keys := make([]*ds.Key, len(nativeKeys)) | 469 keys := make([]*ds.Key, len(nativeKeys)) |
| 457 toks := make([]ds.KeyTok, 1) | 470 toks := make([]ds.KeyTok, 1) |
| 471 |
| 472 kc := bds.kc |
| 458 for i, nativeKey := range nativeKeys { | 473 for i, nativeKey := range nativeKeys { |
| 459 toks = toks[:0] | 474 toks = toks[:0] |
| 460 cur := nativeKey | 475 cur := nativeKey |
| 461 for { | 476 for { |
| 462 toks = append(toks, ds.KeyTok{Kind: cur.Kind(), IntID: c
ur.ID(), StringID: cur.Name()}) | 477 toks = append(toks, ds.KeyTok{Kind: cur.Kind(), IntID: c
ur.ID(), StringID: cur.Name()}) |
| 463 cur = cur.Parent() | 478 cur = cur.Parent() |
| 464 if cur == nil { | 479 if cur == nil { |
| 465 break | 480 break |
| 466 } | 481 } |
| 467 } | 482 } |
| 468 | 483 |
| 469 // Reverse "toks" so we have ancestor-to-child lineage. | 484 // Reverse "toks" so we have ancestor-to-child lineage. |
| 470 for i := 0; i < len(toks)/2; i++ { | 485 for i := 0; i < len(toks)/2; i++ { |
| 471 ri := len(toks) - i - 1 | 486 ri := len(toks) - i - 1 |
| 472 toks[i], toks[ri] = toks[ri], toks[i] | 487 toks[i], toks[ri] = toks[ri], toks[i] |
| 473 } | 488 } |
| 474 » » keys[i] = ds.NewKeyToks(bds.appID, nativeKey.Namespace(), toks) | 489 » » kc.Namespace = nativeKey.Namespace() |
| 490 » » keys[i] = kc.NewKeyToks(toks) |
| 475 } | 491 } |
| 476 return keys | 492 return keys |
| 477 } | 493 } |
| 478 | 494 |
| 479 // nativePropertyLoadSaver is a ds.PropertyMap which implements | 495 // nativePropertyLoadSaver is a ds.PropertyMap which implements |
| 480 // datastore.PropertyLoadSaver. | 496 // datastore.PropertyLoadSaver. |
| 481 // | 497 // |
| 482 // It naturally converts between native and GAE properties and values. | 498 // It naturally converts between native and GAE properties and values. |
| 483 type nativePropertyLoadSaver struct { | 499 type nativePropertyLoadSaver struct { |
| 484 bds *boundDatastore | 500 bds *boundDatastore |
| (...skipping 72 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 557 case datastore.ErrNoSuchEntity: | 573 case datastore.ErrNoSuchEntity: |
| 558 return ds.ErrNoSuchEntity | 574 return ds.ErrNoSuchEntity |
| 559 case datastore.ErrConcurrentTransaction: | 575 case datastore.ErrConcurrentTransaction: |
| 560 return ds.ErrConcurrentTransaction | 576 return ds.ErrConcurrentTransaction |
| 561 case datastore.ErrInvalidKey: | 577 case datastore.ErrInvalidKey: |
| 562 return ds.ErrInvalidKey | 578 return ds.ErrInvalidKey |
| 563 default: | 579 default: |
| 564 return err | 580 return err |
| 565 } | 581 } |
| 566 } | 582 } |
| OLD | NEW |